* rework reference counting in the client libraries: now refcounting goes

  strictly "one-way" - the "bigger" object refcounts the "smaller" one, never the
  other way round. 

* when registering for a reply packet in pdispatch, specify a function that is
  called when the pdispatch object is destroyed but the reply hasn't yet been
  recieved.

* move prototype of pa_free_cb from stream.h to def.h 



git-svn-id: file:///home/lennart/svn/public/pulseaudio/trunk@794 fefdeb5f-60dc-0310-8127-8f9354f1896f

12 files changed, 175 insertions, 116 deletions

diff --git a/src/modules/module-tunnel.c b/src/modules/module-tunnel.c
index b68bc485..abfa68a4 100644
--- a/src/modules/module-tunnel.c
+++ b/src/modules/module-tunnel.c
@@ -356,7 +356,7 @@ static void request_latency(struct userdata *u) {
     pa_tagstruct_put_timeval(t, &now);
     
     pa_pstream_send_tagstruct(u->pstream, t);
-    pa_pdispatch_register_reply(u->pdispatch, tag, DEFAULT_TIMEOUT, stream_get_latency_callback, u);
+    pa_pdispatch_register_reply(u->pdispatch, tag, DEFAULT_TIMEOUT, stream_get_latency_callback, u, NULL);
 }
 
 static void stream_get_info_callback(pa_pdispatch *pd, uint32_t command, PA_GCC_UNUSED uint32_t tag, pa_tagstruct *t, void *userdata) {
@@ -449,7 +449,7 @@ static void request_info(struct userdata *u) {
 #endif
 
     pa_pstream_send_tagstruct(u->pstream, t);
-    pa_pdispatch_register_reply(u->pdispatch, tag, DEFAULT_TIMEOUT, stream_get_info_callback, u);
+    pa_pdispatch_register_reply(u->pdispatch, tag, DEFAULT_TIMEOUT, stream_get_info_callback, u, NULL);
 }
 
 static void start_subscribe(struct userdata *u) {
@@ -580,7 +580,7 @@ static void setup_complete_callback(pa_pdispatch *pd, uint32_t command, uint32_t
 #endif
     
     pa_pstream_send_tagstruct(u->pstream, reply);
-    pa_pdispatch_register_reply(u->pdispatch, tag, DEFAULT_TIMEOUT, create_stream_callback, u);
+    pa_pdispatch_register_reply(u->pdispatch, tag, DEFAULT_TIMEOUT, create_stream_callback, u, NULL);
 }
 
 static void pstream_die_callback(pa_pstream *p, void *userdata) {
@@ -647,7 +647,7 @@ static void on_connection(pa_socket_client *sc, pa_iochannel *io, void *userdata
     pa_tagstruct_putu32(t, PA_PROTOCOL_VERSION);
     pa_tagstruct_put_arbitrary(t, u->auth_cookie, sizeof(u->auth_cookie));
     pa_pstream_send_tagstruct(u->pstream, t);
-    pa_pdispatch_register_reply(u->pdispatch, tag, DEFAULT_TIMEOUT, setup_complete_callback, u);
+    pa_pdispatch_register_reply(u->pdispatch, tag, DEFAULT_TIMEOUT, setup_complete_callback, u, NULL);
     
 }
 
diff --git a/src/polyp/context.c b/src/polyp/context.c
index 1e78f97f..c9a65847 100644
--- a/src/polyp/context.c
+++ b/src/polyp/context.c
@@ -162,7 +162,7 @@ static void context_free(pa_context *c) {
 
     while (c->streams)
         pa_stream_set_state(c->streams, PA_STREAM_TERMINATED);
-    
+
     if (c->client)
         pa_socket_client_unref(c->client);
     if (c->pdispatch)
@@ -218,6 +218,10 @@ void pa_context_set_state(pa_context *c, pa_context_state_t st) {
 
     pa_context_ref(c);
 
+    c->state = st;
+    if (c->state_callback)
+        c->state_callback(c, c->state_userdata);
+
     if (st == PA_CONTEXT_FAILED || st == PA_CONTEXT_TERMINATED) {
         pa_stream *s;
         
@@ -244,10 +248,6 @@ void pa_context_set_state(pa_context *c, pa_context_state_t st) {
         c->client = NULL;
     }
 
-    c->state = st;
-    if (c->state_callback)
-        c->state_callback(c, c->state_userdata);
-
     pa_context_unref(c);
 }
 
@@ -383,7 +383,7 @@ static void setup_complete_callback(pa_pdispatch *pd, uint32_t command, uint32_t
             reply = pa_tagstruct_command(c, PA_COMMAND_SET_CLIENT_NAME, &tag);
             pa_tagstruct_puts(reply, c->name);
             pa_pstream_send_tagstruct(c->pstream, reply);
-            pa_pdispatch_register_reply(c->pdispatch, tag, DEFAULT_TIMEOUT, setup_complete_callback, c);
+            pa_pdispatch_register_reply(c->pdispatch, tag, DEFAULT_TIMEOUT, setup_complete_callback, c, NULL);
 
             pa_context_set_state(c, PA_CONTEXT_SETTING_NAME);
             break;
@@ -429,7 +429,7 @@ static void setup_context(pa_context *c, pa_iochannel *io) {
     pa_tagstruct_putu32(t, PA_PROTOCOL_VERSION);
     pa_tagstruct_put_arbitrary(t, c->conf->cookie, sizeof(c->conf->cookie));
     pa_pstream_send_tagstruct_with_creds(c->pstream, t, 1);
-    pa_pdispatch_register_reply(c->pdispatch, tag, DEFAULT_TIMEOUT, setup_complete_callback, c);
+    pa_pdispatch_register_reply(c->pdispatch, tag, DEFAULT_TIMEOUT, setup_complete_callback, c, NULL);
 
     pa_context_set_state(c, PA_CONTEXT_AUTHORIZING);
 
@@ -817,7 +817,9 @@ void pa_context_simple_ack_callback(pa_pdispatch *pd, uint32_t command, PA_GCC_U
     assert(pd);
     assert(o);
     assert(o->ref >= 1);
-    assert(o->context);
+
+    if (!o->context)
+        goto finish;
 
     if (command != PA_COMMAND_REPLY) {
         if (pa_context_handle_error(o->context, command, t) < 0)
@@ -853,7 +855,7 @@ pa_operation* pa_context_exit_daemon(pa_context *c, pa_context_success_cb_t cb,
 
     t = pa_tagstruct_command(c, PA_COMMAND_EXIT, &tag);
     pa_pstream_send_tagstruct(c->pstream, t);
-    pa_pdispatch_register_reply(c->pdispatch, tag, DEFAULT_TIMEOUT, pa_context_simple_ack_callback, pa_operation_ref(o));
+    pa_pdispatch_register_reply(c->pdispatch, tag, DEFAULT_TIMEOUT, pa_context_simple_ack_callback, pa_operation_ref(o), (pa_free_cb_t) pa_operation_unref);
 
     return o;
 }
@@ -872,7 +874,7 @@ pa_operation* pa_context_send_simple_command(pa_context *c, uint32_t command, pa
 
     t = pa_tagstruct_command(c, command, &tag);
     pa_pstream_send_tagstruct(c->pstream, t);
-    pa_pdispatch_register_reply(c->pdispatch, tag, DEFAULT_TIMEOUT, internal_cb, pa_operation_ref(o));
+    pa_pdispatch_register_reply(c->pdispatch, tag, DEFAULT_TIMEOUT, internal_cb, pa_operation_ref(o), (pa_free_cb_t) pa_operation_unref);
 
     return o;
 }
@@ -892,7 +894,7 @@ pa_operation* pa_context_set_default_sink(pa_context *c, const char *name, pa_co
     t = pa_tagstruct_command(c, PA_COMMAND_SET_DEFAULT_SINK, &tag);
     pa_tagstruct_puts(t, name);
     pa_pstream_send_tagstruct(c->pstream, t);
-    pa_pdispatch_register_reply(c->pdispatch, tag, DEFAULT_TIMEOUT,  pa_context_simple_ack_callback, pa_operation_ref(o));
+    pa_pdispatch_register_reply(c->pdispatch, tag, DEFAULT_TIMEOUT,  pa_context_simple_ack_callback, pa_operation_ref(o), (pa_free_cb_t) pa_operation_unref);
 
     return o;
 }
@@ -912,7 +914,7 @@ pa_operation* pa_context_set_default_source(pa_context *c, const char *name, pa_
     t = pa_tagstruct_command(c, PA_COMMAND_SET_DEFAULT_SOURCE, &tag);
     pa_tagstruct_puts(t, name);
     pa_pstream_send_tagstruct(c->pstream, t);
-    pa_pdispatch_register_reply(c->pdispatch, tag, DEFAULT_TIMEOUT,  pa_context_simple_ack_callback, pa_operation_ref(o));
+    pa_pdispatch_register_reply(c->pdispatch, tag, DEFAULT_TIMEOUT,  pa_context_simple_ack_callback, pa_operation_ref(o), (pa_free_cb_t) pa_operation_unref);
 
     return o;
 }
@@ -939,7 +941,7 @@ pa_operation* pa_context_set_name(pa_context *c, const char *name, pa_context_su
     t = pa_tagstruct_command(c, PA_COMMAND_SET_CLIENT_NAME, &tag);
     pa_tagstruct_puts(t, name);
     pa_pstream_send_tagstruct(c->pstream, t);
-    pa_pdispatch_register_reply(c->pdispatch, tag, DEFAULT_TIMEOUT,  pa_context_simple_ack_callback, pa_operation_ref(o));
+    pa_pdispatch_register_reply(c->pdispatch, tag, DEFAULT_TIMEOUT,  pa_context_simple_ack_callback, pa_operation_ref(o), (pa_free_cb_t) pa_operation_unref);
 
     return o;
 }
diff --git a/src/polyp/def.h b/src/polyp/def.h
index edada380..57997163 100644
--- a/src/polyp/def.h
+++ b/src/polyp/def.h
@@ -304,6 +304,9 @@ typedef enum pa_source_flags {
     PA_SOURCE_LATENCY = 2          /**< Supports latency querying */
 } pa_source_flags_t;
 
+/** A generic free() like callback prototype */
+typedef void (*pa_free_cb_t)(void *p);
+
 PA_C_DECL_END
 
 #endif
diff --git a/src/polyp/internal.h b/src/polyp/internal.h
index 2bec0294..f41744ef 100644
--- a/src/polyp/internal.h
+++ b/src/polyp/internal.h
@@ -159,6 +159,7 @@ struct pa_operation {
     int ref;
     pa_context *context;
     pa_stream *stream;
+    
     PA_LLIST_FIELDS(pa_operation);
 
     pa_operation_state_t state;
diff --git a/src/polyp/introspect.c b/src/polyp/introspect.c
index ea6133e1..9d002669 100644
--- a/src/polyp/introspect.c
+++ b/src/polyp/introspect.c
@@ -43,7 +43,9 @@ static void context_stat_callback(pa_pdispatch *pd, uint32_t command, PA_GCC_UNU
     assert(pd);
     assert(o);
     assert(o->ref >= 1);
-    assert(o->context);
+
+    if (!o->context)
+        goto finish;
 
     if (command != PA_COMMAND_REPLY) {
         if (pa_context_handle_error(o->context, command, t) < 0)
@@ -83,7 +85,9 @@ static void context_get_server_info_callback(pa_pdispatch *pd, uint32_t command,
     assert(pd);
     assert(o);
     assert(o->ref >= 1);
-    assert(o->context);
+    
+    if (!o->context)
+        goto finish;
 
     if (command != PA_COMMAND_REPLY) {
         if (pa_context_handle_error(o->context, command, t) < 0)
@@ -127,8 +131,10 @@ static void context_get_sink_info_callback(pa_pdispatch *pd, uint32_t command, P
     assert(pd);
     assert(o);
     assert(o->ref >= 1);
-    assert(o->context);
 
+    if (!o->context)
+        goto finish;
+    
     if (command != PA_COMMAND_REPLY) {
         if (pa_context_handle_error(o->context, command, t) < 0)
             goto finish;
@@ -198,7 +204,7 @@ pa_operation* pa_context_get_sink_info_by_index(pa_context *c, uint32_t idx, pa_
     pa_tagstruct_putu32(t, idx);
     pa_tagstruct_puts(t, NULL);
     pa_pstream_send_tagstruct(c->pstream, t);
-    pa_pdispatch_register_reply(c->pdispatch, tag, DEFAULT_TIMEOUT, context_get_sink_info_callback, pa_operation_ref(o));
+    pa_pdispatch_register_reply(c->pdispatch, tag, DEFAULT_TIMEOUT, context_get_sink_info_callback, pa_operation_ref(o), (pa_free_cb_t) pa_operation_unref);
 
     return o;
 }
@@ -221,7 +227,7 @@ pa_operation* pa_context_get_sink_info_by_name(pa_context *c, const char *name,
     pa_tagstruct_putu32(t, PA_INVALID_INDEX);
     pa_tagstruct_puts(t, name);
     pa_pstream_send_tagstruct(c->pstream, t);
-    pa_pdispatch_register_reply(c->pdispatch, tag, DEFAULT_TIMEOUT, context_get_sink_info_callback, pa_operation_ref(o));
+    pa_pdispatch_register_reply(c->pdispatch, tag, DEFAULT_TIMEOUT, context_get_sink_info_callback, pa_operation_ref(o), (pa_free_cb_t) pa_operation_unref);
 
     return o;
 }
@@ -235,7 +241,9 @@ static void context_get_source_info_callback(pa_pdispatch *pd, uint32_t command,
     assert(pd);
     assert(o);
     assert(o->ref >= 1);
-    assert(o->context);
+
+    if (!o->context)
+        goto finish;
 
     if (command != PA_COMMAND_REPLY) {
         if (pa_context_handle_error(o->context, command, t) < 0)
@@ -306,7 +314,7 @@ pa_operation* pa_context_get_source_info_by_index(pa_context *c, uint32_t idx, p
     pa_tagstruct_putu32(t, idx);
     pa_tagstruct_puts(t, NULL);
     pa_pstream_send_tagstruct(c->pstream, t);
-    pa_pdispatch_register_reply(c->pdispatch, tag, DEFAULT_TIMEOUT, context_get_source_info_callback, pa_operation_ref(o));
+    pa_pdispatch_register_reply(c->pdispatch, tag, DEFAULT_TIMEOUT, context_get_source_info_callback, pa_operation_ref(o), (pa_free_cb_t) pa_operation_unref);
 
     return o;
 }
@@ -329,9 +337,9 @@ pa_operation* pa_context_get_source_info_by_name(pa_context *c, const char *name
     pa_tagstruct_putu32(t, PA_INVALID_INDEX);
     pa_tagstruct_puts(t, name);
     pa_pstream_send_tagstruct(c->pstream, t);
-    pa_pdispatch_register_reply(c->pdispatch, tag, DEFAULT_TIMEOUT, context_get_source_info_callback, o);
+    pa_pdispatch_register_reply(c->pdispatch, tag, DEFAULT_TIMEOUT, context_get_source_info_callback, pa_operation_ref(o), (pa_free_cb_t) pa_operation_unref);
 
-    return pa_operation_ref(o);
+    return o;
 }
 
 /*** Client info ***/
@@ -343,7 +351,9 @@ static void context_get_client_info_callback(pa_pdispatch *pd, uint32_t command,
     assert(pd);
     assert(o);
     assert(o->ref >= 1);
-    assert(o->context);
+
+    if (!o->context)
+        goto finish;
 
     if (command != PA_COMMAND_REPLY) {
         if (pa_context_handle_error(o->context, command, t) < 0)
@@ -397,7 +407,7 @@ pa_operation* pa_context_get_client_info(pa_context *c, uint32_t idx, pa_client_
     t = pa_tagstruct_command(c, PA_COMMAND_GET_CLIENT_INFO, &tag);
     pa_tagstruct_putu32(t, idx);
     pa_pstream_send_tagstruct(c->pstream, t);
-    pa_pdispatch_register_reply(c->pdispatch, tag, DEFAULT_TIMEOUT, context_get_client_info_callback, pa_operation_ref(o));
+    pa_pdispatch_register_reply(c->pdispatch, tag, DEFAULT_TIMEOUT, context_get_client_info_callback, pa_operation_ref(o), (pa_free_cb_t) pa_operation_unref);
 
     return o;
 }
@@ -415,7 +425,9 @@ static void context_get_module_info_callback(pa_pdispatch *pd, uint32_t command,
     assert(pd);
     assert(o);
     assert(o->ref >= 1);
-    assert(o->context);
+
+    if (!o->context)
+        goto finish;
     
     if (command != PA_COMMAND_REPLY) {
         if (pa_context_handle_error(o->context, command, t) < 0)
@@ -470,7 +482,7 @@ pa_operation* pa_context_get_module_info(pa_context *c, uint32_t idx, pa_module_
     t = pa_tagstruct_command(c, PA_COMMAND_GET_MODULE_INFO, &tag);
     pa_tagstruct_putu32(t, idx);
     pa_pstream_send_tagstruct(c->pstream, t);
-    pa_pdispatch_register_reply(c->pdispatch, tag, DEFAULT_TIMEOUT, context_get_module_info_callback, pa_operation_ref(o));
+    pa_pdispatch_register_reply(c->pdispatch, tag, DEFAULT_TIMEOUT, context_get_module_info_callback, pa_operation_ref(o), (pa_free_cb_t) pa_operation_unref);
 
     return o;
 }
@@ -488,8 +500,10 @@ static void context_get_sink_input_info_callback(pa_pdispatch *pd, uint32_t comm
     assert(pd);
     assert(o);
     assert(o->ref >= 1);
-    assert(o->context);
 
+    if (!o->context)
+        goto finish;
+    
     if (command != PA_COMMAND_REPLY) {
         if (pa_context_handle_error(o->context, command, t) < 0)
             goto finish;
@@ -551,7 +565,7 @@ pa_operation* pa_context_get_sink_input_info(pa_context *c, uint32_t idx, pa_sin
     t = pa_tagstruct_command(c, PA_COMMAND_GET_SINK_INPUT_INFO, &tag);
     pa_tagstruct_putu32(t, idx);
     pa_pstream_send_tagstruct(c->pstream, t);
-    pa_pdispatch_register_reply(c->pdispatch, tag, DEFAULT_TIMEOUT, context_get_sink_input_info_callback, pa_operation_ref(o));
+    pa_pdispatch_register_reply(c->pdispatch, tag, DEFAULT_TIMEOUT, context_get_sink_input_info_callback, pa_operation_ref(o), (pa_free_cb_t) pa_operation_unref);
 
     return o;
 }
@@ -569,7 +583,9 @@ static void context_get_source_output_info_callback(pa_pdispatch *pd, uint32_t c
     assert(pd);
     assert(o);
     assert(o->ref >= 1);
-    assert(o->context);
+
+    if (!o->context)
+        goto finish;
 
     if (command != PA_COMMAND_REPLY) {
         if (pa_context_handle_error(o->context, command, t) < 0)
@@ -631,7 +647,7 @@ pa_operation* pa_context_get_source_output_info(pa_context *c, uint32_t idx, pa_
     t = pa_tagstruct_command(c, PA_COMMAND_GET_SOURCE_OUTPUT_INFO, &tag);
     pa_tagstruct_putu32(t, idx);
     pa_pstream_send_tagstruct(c->pstream, t);
-    pa_pdispatch_register_reply(c->pdispatch, tag, DEFAULT_TIMEOUT, context_get_source_output_info_callback, pa_operation_ref(o));
+    pa_pdispatch_register_reply(c->pdispatch, tag, DEFAULT_TIMEOUT, context_get_source_output_info_callback, pa_operation_ref(o), (pa_free_cb_t) pa_operation_unref);
 
     return o;
 }
@@ -661,7 +677,7 @@ pa_operation* pa_context_set_sink_volume_by_index(pa_context *c, uint32_t idx, c
     pa_tagstruct_puts(t, NULL);
     pa_tagstruct_put_cvolume(t, volume);
     pa_pstream_send_tagstruct(c->pstream, t);
-    pa_pdispatch_register_reply(c->pdispatch, tag, DEFAULT_TIMEOUT, pa_context_simple_ack_callback, pa_operation_ref(o));
+    pa_pdispatch_register_reply(c->pdispatch, tag, DEFAULT_TIMEOUT, pa_context_simple_ack_callback, pa_operation_ref(o), (pa_free_cb_t) pa_operation_unref);
 
     return o;
 }
@@ -687,7 +703,7 @@ pa_operation* pa_context_set_sink_volume_by_name(pa_context *c, const char *name
     pa_tagstruct_puts(t, name);
     pa_tagstruct_put_cvolume(t, volume);
     pa_pstream_send_tagstruct(c->pstream, t);
-    pa_pdispatch_register_reply(c->pdispatch, tag, DEFAULT_TIMEOUT, pa_context_simple_ack_callback, pa_operation_ref(o));
+    pa_pdispatch_register_reply(c->pdispatch, tag, DEFAULT_TIMEOUT, pa_context_simple_ack_callback, pa_operation_ref(o), (pa_free_cb_t) pa_operation_unref);
 
     return o;
 }
@@ -709,7 +725,7 @@ pa_operation* pa_context_set_sink_mute_by_index(pa_context *c, uint32_t idx, int
     pa_tagstruct_puts(t, NULL);
     pa_tagstruct_put_boolean(t, mute);
     pa_pstream_send_tagstruct(c->pstream, t);
-    pa_pdispatch_register_reply(c->pdispatch, tag, DEFAULT_TIMEOUT, pa_context_simple_ack_callback, pa_operation_ref(o));
+    pa_pdispatch_register_reply(c->pdispatch, tag, DEFAULT_TIMEOUT, pa_context_simple_ack_callback, pa_operation_ref(o), (pa_free_cb_t) pa_operation_unref);
 
     return o;
 }
@@ -733,7 +749,7 @@ pa_operation* pa_context_set_sink_mute_by_name(pa_context *c, const char *name,
     pa_tagstruct_puts(t, name);
     pa_tagstruct_put_boolean(t, mute);
     pa_pstream_send_tagstruct(c->pstream, t);
-    pa_pdispatch_register_reply(c->pdispatch, tag, DEFAULT_TIMEOUT, pa_context_simple_ack_callback, pa_operation_ref(o));
+    pa_pdispatch_register_reply(c->pdispatch, tag, DEFAULT_TIMEOUT, pa_context_simple_ack_callback, pa_operation_ref(o), (pa_free_cb_t) pa_operation_unref);
 
     return o;
 }
@@ -757,7 +773,7 @@ pa_operation* pa_context_set_sink_input_volume(pa_context *c, uint32_t idx, cons
     pa_tagstruct_putu32(t, idx);
     pa_tagstruct_put_cvolume(t, volume);
     pa_pstream_send_tagstruct(c->pstream, t);
-    pa_pdispatch_register_reply(c->pdispatch, tag, DEFAULT_TIMEOUT, pa_context_simple_ack_callback, pa_operation_ref(o));
+    pa_pdispatch_register_reply(c->pdispatch, tag, DEFAULT_TIMEOUT, pa_context_simple_ack_callback, pa_operation_ref(o), (pa_free_cb_t) pa_operation_unref);
 
     return o;
 }
@@ -781,7 +797,7 @@ pa_operation* pa_context_set_source_volume_by_index(pa_context *c, uint32_t idx,
     pa_tagstruct_puts(t, NULL);
     pa_tagstruct_put_cvolume(t, volume);
     pa_pstream_send_tagstruct(c->pstream, t);
-    pa_pdispatch_register_reply(c->pdispatch, tag, DEFAULT_TIMEOUT, pa_context_simple_ack_callback, pa_operation_ref(o));
+    pa_pdispatch_register_reply(c->pdispatch, tag, DEFAULT_TIMEOUT, pa_context_simple_ack_callback, pa_operation_ref(o), (pa_free_cb_t) pa_operation_unref);
 
     return o;
 }
@@ -807,7 +823,7 @@ pa_operation* pa_context_set_source_volume_by_name(pa_context *c, const char *na
     pa_tagstruct_puts(t, name);
     pa_tagstruct_put_cvolume(t, volume);
     pa_pstream_send_tagstruct(c->pstream, t);
-    pa_pdispatch_register_reply(c->pdispatch, tag, DEFAULT_TIMEOUT, pa_context_simple_ack_callback, pa_operation_ref(o));
+    pa_pdispatch_register_reply(c->pdispatch, tag, DEFAULT_TIMEOUT, pa_context_simple_ack_callback, pa_operation_ref(o), (pa_free_cb_t) pa_operation_unref);
 
     return o;
 }
@@ -829,7 +845,7 @@ pa_operation* pa_context_set_source_mute_by_index(pa_context *c, uint32_t idx, i
     pa_tagstruct_puts(t, NULL);
     pa_tagstruct_put_boolean(t, mute);
     pa_pstream_send_tagstruct(c->pstream, t);
-    pa_pdispatch_register_reply(c->pdispatch, tag, DEFAULT_TIMEOUT, pa_context_simple_ack_callback, pa_operation_ref(o));
+    pa_pdispatch_register_reply(c->pdispatch, tag, DEFAULT_TIMEOUT, pa_context_simple_ack_callback, pa_operation_ref(o), (pa_free_cb_t) pa_operation_unref);
 
     return o;
 }
@@ -853,7 +869,7 @@ pa_operation* pa_context_set_source_mute_by_name(pa_context *c, const char *name
     pa_tagstruct_puts(t, name);
     pa_tagstruct_put_boolean(t, mute);
     pa_pstream_send_tagstruct(c->pstream, t);
-    pa_pdispatch_register_reply(c->pdispatch, tag, DEFAULT_TIMEOUT, pa_context_simple_ack_callback, pa_operation_ref(o));
+    pa_pdispatch_register_reply(c->pdispatch, tag, DEFAULT_TIMEOUT, pa_context_simple_ack_callback, pa_operation_ref(o), (pa_free_cb_t) pa_operation_unref);
 
     return o;
 }
@@ -867,8 +883,10 @@ static void context_get_sample_info_callback(pa_pdispatch *pd, uint32_t command,
     assert(pd);
     assert(o);
     assert(o->ref >= 1);
-    assert(o->context);
 
+    if (!o->context)
+        goto finish;
+    
     if (command != PA_COMMAND_REPLY) {
         if (pa_context_handle_error(o->context, command, t) < 0)
             goto finish;
@@ -928,7 +946,7 @@ pa_operation* pa_context_get_sample_info_by_name(pa_context *c, const char *name
     pa_tagstruct_putu32(t, PA_INVALID_INDEX);
     pa_tagstruct_puts(t, name);
     pa_pstream_send_tagstruct(c->pstream, t);
-    pa_pdispatch_register_reply(c->pdispatch, tag, DEFAULT_TIMEOUT, context_get_sample_info_callback, pa_operation_ref(o));
+    pa_pdispatch_register_reply(c->pdispatch, tag, DEFAULT_TIMEOUT, context_get_sample_info_callback, pa_operation_ref(o), (pa_free_cb_t) pa_operation_unref);
 
     return o;
 }
@@ -951,7 +969,7 @@ pa_operation* pa_context_get_sample_info_by_index(pa_context *c, uint32_t idx, p
     pa_tagstruct_putu32(t, idx);
     pa_tagstruct_puts(t, NULL);
     pa_pstream_send_tagstruct(c->pstream, t);
-    pa_pdispatch_register_reply(c->pdispatch, tag, DEFAULT_TIMEOUT, context_get_sample_info_callback, pa_operation_ref(o));
+    pa_pdispatch_register_reply(c->pdispatch, tag, DEFAULT_TIMEOUT, context_get_sample_info_callback, pa_operation_ref(o), (pa_free_cb_t) pa_operation_unref);
 
     return o;
 }
@@ -976,7 +994,7 @@ static pa_operation* command_kill(pa_context *c, uint32_t command, uint32_t idx,
     t = pa_tagstruct_command(c, command, &tag);
     pa_tagstruct_putu32(t, idx);
     pa_pstream_send_tagstruct(c->pstream, t);
-    pa_pdispatch_register_reply(c->pdispatch, tag, DEFAULT_TIMEOUT, pa_context_simple_ack_callback, pa_operation_ref(o));
+    pa_pdispatch_register_reply(c->pdispatch, tag, DEFAULT_TIMEOUT, pa_context_simple_ack_callback, pa_operation_ref(o), (pa_free_cb_t) pa_operation_unref);
 
     return o;
 }
@@ -1000,7 +1018,9 @@ static void context_index_callback(pa_pdispatch *pd, uint32_t command, PA_GCC_UN
     assert(pd);
     assert(o);
     assert(o->ref >= 1);
-    assert(o->context);
+
+    if (!o->context)
+        goto finish;
     
     if (command != PA_COMMAND_REPLY) {
         if (pa_context_handle_error(o->context, command, t) < 0)
@@ -1041,7 +1061,7 @@ pa_operation* pa_context_load_module(pa_context *c, const char*name, const char
     pa_tagstruct_puts(t, name);
     pa_tagstruct_puts(t, argument);
     pa_pstream_send_tagstruct(c->pstream, t);
-    pa_pdispatch_register_reply(c->pdispatch, tag, DEFAULT_TIMEOUT, context_index_callback, pa_operation_ref(o));
+    pa_pdispatch_register_reply(c->pdispatch, tag, DEFAULT_TIMEOUT, context_index_callback, pa_operation_ref(o), (pa_free_cb_t) pa_operation_unref);
 
     return o;
 }
@@ -1059,7 +1079,9 @@ static void context_get_autoload_info_callback(pa_pdispatch *pd, uint32_t comman
     assert(pd);
     assert(o);
     assert(o->ref >= 1);
-    assert(o->context);
+
+    if (!o->context)
+        goto finish;
 
     if (command != PA_COMMAND_REPLY) {
         if (pa_context_handle_error(o->context, command, t) < 0)
@@ -1116,7 +1138,7 @@ pa_operation* pa_context_get_autoload_info_by_name(pa_context *c, const char *na
     pa_tagstruct_puts(t, name);
     pa_tagstruct_putu32(t, type);
     pa_pstream_send_tagstruct(c->pstream, t);
-    pa_pdispatch_register_reply(c->pdispatch, tag, DEFAULT_TIMEOUT, context_get_autoload_info_callback, pa_operation_ref(o));
+    pa_pdispatch_register_reply(c->pdispatch, tag, DEFAULT_TIMEOUT, context_get_autoload_info_callback, pa_operation_ref(o), (pa_free_cb_t) pa_operation_unref);
 
     return o;
 }
@@ -1138,7 +1160,7 @@ pa_operation* pa_context_get_autoload_info_by_index(pa_context *c, uint32_t idx,
     t = pa_tagstruct_command(c, PA_COMMAND_GET_AUTOLOAD_INFO, &tag);
     pa_tagstruct_putu32(t, idx);
     pa_pstream_send_tagstruct(c->pstream, t);
-    pa_pdispatch_register_reply(c->pdispatch, tag, DEFAULT_TIMEOUT, context_get_autoload_info_callback, pa_operation_ref(o));
+    pa_pdispatch_register_reply(c->pdispatch, tag, DEFAULT_TIMEOUT, context_get_autoload_info_callback, pa_operation_ref(o), (pa_free_cb_t) pa_operation_unref);
 
     return o;
 }
@@ -1168,7 +1190,7 @@ pa_operation* pa_context_add_autoload(pa_context *c, const char *name, pa_autolo
     pa_tagstruct_puts(t, module);
     pa_tagstruct_puts(t, argument);
     pa_pstream_send_tagstruct(c->pstream, t);
-    pa_pdispatch_register_reply(c->pdispatch, tag, DEFAULT_TIMEOUT, context_index_callback, pa_operation_ref(o));
+    pa_pdispatch_register_reply(c->pdispatch, tag, DEFAULT_TIMEOUT, context_index_callback, pa_operation_ref(o), (pa_free_cb_t) pa_operation_unref);
 
     return o;
 }
@@ -1191,7 +1213,7 @@ pa_operation* pa_context_remove_autoload_by_name(pa_context *c, const char *name
     pa_tagstruct_puts(t, name);
     pa_tagstruct_putu32(t, type);
     pa_pstream_send_tagstruct(c->pstream, t);
-    pa_pdispatch_register_reply(c->pdispatch, tag, DEFAULT_TIMEOUT, pa_context_simple_ack_callback, pa_operation_ref(o));
+    pa_pdispatch_register_reply(c->pdispatch, tag, DEFAULT_TIMEOUT, pa_context_simple_ack_callback, pa_operation_ref(o), (pa_free_cb_t) pa_operation_unref);
 
     return o;
 }
@@ -1212,7 +1234,7 @@ pa_operation* pa_context_remove_autoload_by_index(pa_context *c, uint32_t idx, p
     t = pa_tagstruct_command(c, PA_COMMAND_REMOVE_AUTOLOAD, &tag);
     pa_tagstruct_putu32(t, idx);
     pa_pstream_send_tagstruct(c->pstream, t);
-    pa_pdispatch_register_reply(c->pdispatch, tag, DEFAULT_TIMEOUT, pa_context_simple_ack_callback, pa_operation_ref(o));
+    pa_pdispatch_register_reply(c->pdispatch, tag, DEFAULT_TIMEOUT, pa_context_simple_ack_callback, pa_operation_ref(o), (pa_free_cb_t) pa_operation_unref);
 
     return o;
 }
diff --git a/src/polyp/operation.c b/src/polyp/operation.c
index 0216888c..1c0cb99f 100644
--- a/src/polyp/operation.c
+++ b/src/polyp/operation.c
@@ -28,78 +28,89 @@
 #include <polypcore/xmalloc.h>
 
 #include "internal.h"
-
 #include "operation.h"
 
 pa_operation *pa_operation_new(pa_context *c, pa_stream *s, pa_operation_cb_t cb, void *userdata) {
     pa_operation *o;
     assert(c);
 
-    o = pa_xmalloc(sizeof(pa_operation));
+    o = pa_xnew(pa_operation, 1);
     o->ref = 1;
-    o->context = pa_context_ref(c);
-    o->stream = s ? pa_stream_ref(s) : NULL;
+    o->context = c;
+    o->stream = s;
 
     o->state = PA_OPERATION_RUNNING;
     o->callback = cb;
     o->userdata = userdata;
 
-    PA_LLIST_PREPEND(pa_operation, o->context->operations, o);
-    return pa_operation_ref(o);
+    /* Refcounting is strictly one-way: from the "bigger" to the "smaller" object. */
+    PA_LLIST_PREPEND(pa_operation, c->operations, o);
+    pa_operation_ref(o);
+    
+    return o;
 }
 
 pa_operation *pa_operation_ref(pa_operation *o) {
-    assert(o && o->ref >= 1);
+    assert(o);
+    assert(o->ref >= 1);
+    
     o->ref++;
     return o;
 }
 
 void pa_operation_unref(pa_operation *o) {
-    assert(o && o->ref >= 1);
+    assert(o);
+    assert(o->ref >= 1);
 
     if ((--(o->ref)) == 0) {
         assert(!o->context);
         assert(!o->stream);
-        free(o);
+        pa_xfree(o);
     }
 }
 
 static void operation_set_state(pa_operation *o, pa_operation_state_t st) {
-    assert(o && o->ref >= 1);
+    assert(o);
+    assert(o->ref >= 1);
 
     if (st == o->state)
         return;
 
-    if (!o->context)
-        return;
-
     o->state = st;
 
     if ((o->state == PA_OPERATION_DONE) || (o->state == PA_OPERATION_CANCELED)) {
-        PA_LLIST_REMOVE(pa_operation, o->context->operations, o);
-        pa_context_unref(o->context);
-        if (o->stream)
-            pa_stream_unref(o->stream);
+        
+        if (o->context) {
+            assert(o->ref >= 2);
+            
+            PA_LLIST_REMOVE(pa_operation, o->context->operations, o);
+            pa_operation_unref(o);
+        }
+        
         o->context = NULL;
         o->stream = NULL;
         o->callback = NULL;
         o->userdata = NULL;
-
-        pa_operation_unref(o);
     }
 }
 
 void pa_operation_cancel(pa_operation *o) {
-    assert(o && o->ref >= 1);
+    assert(o);
+    assert(o->ref >= 1);
+    
     operation_set_state(o, PA_OPERATION_CANCELED);
 }
 
 void pa_operation_done(pa_operation *o) {
-    assert(o && o->ref >= 1);
+    assert(o);
+    assert(o->ref >= 1);
+    
     operation_set_state(o, PA_OPERATION_DONE);
 }
 
 pa_operation_state_t pa_operation_get_state(pa_operation *o) {
-    assert(o && o->ref >= 1);
+    assert(o);
+    assert(o->ref >= 1);
+
     return o->state;
 }
diff --git a/src/polyp/scache.c b/src/polyp/scache.c
index 2b9e9c8f..22d8a545 100644
--- a/src/polyp/scache.c
+++ b/src/polyp/scache.c
@@ -53,7 +53,7 @@ int pa_stream_connect_upload(pa_stream *s, size_t length) {
     pa_tagstruct_put_channel_map(t, &s->channel_map);
     pa_tagstruct_putu32(t, length);
     pa_pstream_send_tagstruct(s->context->pstream, t);
-    pa_pdispatch_register_reply(s->context->pdispatch, tag, DEFAULT_TIMEOUT, pa_create_stream_callback, s);
+    pa_pdispatch_register_reply(s->context->pdispatch, tag, DEFAULT_TIMEOUT, pa_create_stream_callback, s, NULL);
 
     pa_stream_set_state(s, PA_STREAM_CREATING);
     
@@ -74,7 +74,7 @@ int pa_stream_finish_upload(pa_stream *s) {
     t = pa_tagstruct_command(s->context, PA_COMMAND_FINISH_UPLOAD_STREAM, &tag);
     pa_tagstruct_putu32(t, s->channel);
     pa_pstream_send_tagstruct(s->context->pstream, t);
-    pa_pdispatch_register_reply(s->context->pdispatch, tag, DEFAULT_TIMEOUT, pa_stream_disconnect_callback, s);
+    pa_pdispatch_register_reply(s->context->pdispatch, tag, DEFAULT_TIMEOUT, pa_stream_disconnect_callback, s, NULL);
 
     pa_stream_unref(s);
     return 0;
@@ -103,7 +103,7 @@ pa_operation *pa_context_play_sample(pa_context *c, const char *name, const char
     pa_tagstruct_putu32(t, volume);
     pa_tagstruct_puts(t, name);
     pa_pstream_send_tagstruct(c->pstream, t);
-    pa_pdispatch_register_reply(c->pdispatch, tag, DEFAULT_TIMEOUT, pa_context_simple_ack_callback, pa_operation_ref(o));
+    pa_pdispatch_register_reply(c->pdispatch, tag, DEFAULT_TIMEOUT, pa_context_simple_ack_callback, pa_operation_ref(o), (pa_free_cb_t) pa_operation_unref);
 
     return o;
 }
@@ -124,7 +124,7 @@ pa_operation* pa_context_remove_sample(pa_context *c, const char *name, pa_conte
     t = pa_tagstruct_command(c, PA_COMMAND_REMOVE_SAMPLE, &tag);
     pa_tagstruct_puts(t, name);
     pa_pstream_send_tagstruct(c->pstream, t);
-    pa_pdispatch_register_reply(c->pdispatch, tag, DEFAULT_TIMEOUT, pa_context_simple_ack_callback, pa_operation_ref(o));
+    pa_pdispatch_register_reply(c->pdispatch, tag, DEFAULT_TIMEOUT, pa_context_simple_ack_callback, pa_operation_ref(o), (pa_free_cb_t) pa_operation_unref);
 
     return o;
 }
diff --git a/src/polyp/stream.c b/src/polyp/stream.c
index f11ef493..c86a200a 100644
--- a/src/polyp/stream.c
+++ b/src/polyp/stream.c
@@ -106,21 +106,15 @@ pa_stream *pa_stream_new(pa_context *c, const char *name, const pa_sample_spec *
     s->auto_timing_update_event = NULL;
     s->auto_timing_update_requested = 0;
 
+    /* Refcounting is strictly one-way: from the "bigger" to the "smaller" object. */
     PA_LLIST_PREPEND(pa_stream, c->streams, s);
-
-    /* The context and stream will point at each other. We cannot ref count
-       both though since that will create a loop. */
-    pa_context_ref(s->context);
+    pa_stream_ref(s);
 
     return s;
 }
 
 static void stream_free(pa_stream *s) {
-    assert(s && s->context && !s->channel_valid);
-
-    PA_LLIST_REMOVE(pa_stream, s->context->streams, s);
-
-    pa_context_unref(s->context);
+    assert(s && !s->context && !s->channel_valid);
 
     if (s->auto_timing_update_event) {
         assert(s->mainloop);
@@ -186,20 +180,38 @@ void pa_stream_set_state(pa_stream *s, pa_stream_state_t st) {
     pa_stream_ref(s);
 
     s->state = st;
-    
+    if (s->state_callback)
+        s->state_callback(s, s->state_userdata);
+
     if ((st == PA_STREAM_FAILED || st == PA_STREAM_TERMINATED) && s->context) {
+
         /* Detach from context */
+        pa_operation *o, *n;
+
+        /* Unref all operatio object that point to us */
+        for (o = s->context->operations; o; o = n) {
+            n = o->next;
+            
+            if (o->stream == s)
+                pa_operation_cancel(o);
+        }
         
+        /* Drop all outstanding replies for this stream */
+        if (s->context->pdispatch)
+            pa_pdispatch_unregister_reply(s->context->pdispatch, s);
+
         if (s->channel_valid)
             pa_dynarray_put((s->direction == PA_STREAM_PLAYBACK) ? s->context->playback_streams : s->context->record_streams, s->channel, NULL);
+        
+        PA_LLIST_REMOVE(pa_stream, s->context->streams, s);
+        pa_stream_unref(s);
 
         s->channel = 0;
         s->channel_valid = 0;
+ 
+        s->context = NULL;
     }
 
-    if (s->state_callback)
-        s->state_callback(s, s->state_userdata);
-
     pa_stream_unref(s);
 }
 
@@ -513,7 +525,7 @@ static int create_stream(
         pa_tagstruct_putu32(t, s->buffer_attr.fragsize);
 
     pa_pstream_send_tagstruct(s->context->pstream, t);
-    pa_pdispatch_register_reply(s->context->pdispatch, tag, DEFAULT_TIMEOUT, pa_create_stream_callback, s);
+    pa_pdispatch_register_reply(s->context->pdispatch, tag, DEFAULT_TIMEOUT, pa_create_stream_callback, s, NULL);
 
     pa_stream_set_state(s, PA_STREAM_CREATING);
     
@@ -704,9 +716,9 @@ pa_operation * pa_stream_drain(pa_stream *s, pa_stream_success_cb_t cb, void *us
     t = pa_tagstruct_command(s->context, PA_COMMAND_DRAIN_PLAYBACK_STREAM, &tag);
     pa_tagstruct_putu32(t, s->channel);
     pa_pstream_send_tagstruct(s->context->pstream, t);
-    pa_pdispatch_register_reply(s->context->pdispatch, tag, DEFAULT_TIMEOUT, pa_stream_simple_ack_callback, o);
+    pa_pdispatch_register_reply(s->context->pdispatch, tag, DEFAULT_TIMEOUT, pa_stream_simple_ack_callback, pa_operation_ref(o), (pa_free_cb_t) pa_operation_unref);
 
-    return pa_operation_ref(o);
+    return o;
 }
 
 static void stream_get_timing_info_callback(pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata) {
@@ -717,7 +729,9 @@ static void stream_get_timing_info_callback(pa_pdispatch *pd, uint32_t command,
     assert(pd);
     assert(o);
     assert(o->stream);
-    assert(o->context);
+
+    if (!o->context)
+        goto finish;
 
     i = &o->stream->timing_info;
 
@@ -869,7 +883,7 @@ pa_operation* pa_stream_update_timing_info(pa_stream *s, pa_stream_success_cb_t
         /* Check if we could allocate a correction slot. If not, there are too many outstanding queries */
         PA_CHECK_VALIDITY_RETURN_NULL(s->context, !s->write_index_corrections[cidx].valid, PA_ERR_INTERNAL);
     }
-      o = pa_operation_new(s->context, s, (pa_operation_cb_t) cb, userdata);
+    o = pa_operation_new(s->context, s, (pa_operation_cb_t) cb, userdata);
     
     t = pa_tagstruct_command(
             s->context,
@@ -879,7 +893,7 @@ pa_operation* pa_stream_update_timing_info(pa_stream *s, pa_stream_success_cb_t
     pa_tagstruct_put_timeval(t, pa_gettimeofday(&now));
     
     pa_pstream_send_tagstruct(s->context->pstream, t);
-    pa_pdispatch_register_reply(s->context->pdispatch, tag, DEFAULT_TIMEOUT, stream_get_timing_info_callback, o);
+    pa_pdispatch_register_reply(s->context->pdispatch, tag, DEFAULT_TIMEOUT, stream_get_timing_info_callback, pa_operation_ref(o), (pa_free_cb_t) pa_operation_unref);
 
     if (s->direction == PA_STREAM_PLAYBACK) {
         /* Fill in initial correction data */
@@ -893,7 +907,7 @@ pa_operation* pa_stream_update_timing_info(pa_stream *s, pa_stream_success_cb_t
     
 /*     pa_log("requesting update %u\n", tag); */
     
-    return pa_operation_ref(o);
+    return o;
 }
 
 void pa_stream_disconnect_callback(pa_pdispatch *pd, uint32_t command, PA_GCC_UNUSED uint32_t tag, pa_tagstruct *t, void *userdata) {
@@ -941,7 +955,7 @@ int pa_stream_disconnect(pa_stream *s) {
             &tag);
     pa_tagstruct_putu32(t, s->channel);
     pa_pstream_send_tagstruct(s->context->pstream, t);
-    pa_pdispatch_register_reply(s->context->pdispatch, tag, DEFAULT_TIMEOUT, pa_stream_disconnect_callback, s);
+    pa_pdispatch_register_reply(s->context->pdispatch, tag, DEFAULT_TIMEOUT, pa_stream_disconnect_callback, s, NULL);
 
     pa_stream_unref(s);
     return 0;
@@ -993,9 +1007,11 @@ void pa_stream_simple_ack_callback(pa_pdispatch *pd, uint32_t command, PA_GCC_UN
     
     assert(pd);
     assert(o);
-    assert(o->context);
     assert(o->ref >= 1);
 
+    if (!o->context)
+        goto finish;
+    
     if (command != PA_COMMAND_REPLY) {
         if (pa_context_handle_error(o->context, command, t) < 0)
             goto finish;
@@ -1038,12 +1054,12 @@ pa_operation* pa_stream_cork(pa_stream *s, int b, pa_stream_success_cb_t cb, voi
     pa_tagstruct_putu32(t, s->channel);
     pa_tagstruct_put_boolean(t, !!b);
     pa_pstream_send_tagstruct(s->context->pstream, t);
-    pa_pdispatch_register_reply(s->context->pdispatch, tag, DEFAULT_TIMEOUT, pa_stream_simple_ack_callback, o);
+    pa_pdispatch_register_reply(s->context->pdispatch, tag, DEFAULT_TIMEOUT, pa_stream_simple_ack_callback, pa_operation_ref(o), (pa_free_cb_t) pa_operation_unref);
 
     if (s->direction == PA_STREAM_PLAYBACK)
         invalidate_indexes(s, 1, 0);
 
-    return pa_operation_ref(o);
+    return o;
 }
 
 static pa_operation* stream_send_simple_command(pa_stream *s, uint32_t command, pa_stream_success_cb_t cb, void *userdata) {
@@ -1061,9 +1077,9 @@ static pa_operation* stream_send_simple_command(pa_stream *s, uint32_t command,
     t = pa_tagstruct_command(s->context, command, &tag);
     pa_tagstruct_putu32(t, s->channel);
     pa_pstream_send_tagstruct(s->context->pstream, t);
-    pa_pdispatch_register_reply(s->context->pdispatch, tag, DEFAULT_TIMEOUT, pa_stream_simple_ack_callback, o);
+    pa_pdispatch_register_reply(s->context->pdispatch, tag, DEFAULT_TIMEOUT, pa_stream_simple_ack_callback, pa_operation_ref(o), (pa_free_cb_t) pa_operation_unref);
 
-    return pa_operation_ref(o);
+    return o;
 }
 
 pa_operation* pa_stream_flush(pa_stream *s, pa_stream_success_cb_t cb, void *userdata) {
@@ -1136,9 +1152,9 @@ pa_operation* pa_stream_set_name(pa_stream *s, const char *name, pa_stream_succe
     pa_tagstruct_putu32(t, s->channel);
     pa_tagstruct_puts(t, name);
     pa_pstream_send_tagstruct(s->context->pstream, t);
-    pa_pdispatch_register_reply(s->context->pdispatch, tag, DEFAULT_TIMEOUT, pa_stream_simple_ack_callback, o);
+    pa_pdispatch_register_reply(s->context->pdispatch, tag, DEFAULT_TIMEOUT, pa_stream_simple_ack_callback, pa_operation_ref(o), (pa_free_cb_t) pa_operation_unref);
 
-    return pa_operation_ref(o);
+    return o;
 }
 
 int pa_stream_get_time(pa_stream *s, pa_usec_t *r_usec) {
diff --git a/src/polyp/stream.h b/src/polyp/stream.h
index a832fc66..b6522efe 100644
--- a/src/polyp/stream.h
+++ b/src/polyp/stream.h
@@ -267,9 +267,6 @@ typedef struct pa_stream pa_stream;
 /** A generic callback for operation completion */
 typedef void (*pa_stream_success_cb_t) (pa_stream*s, int success, void *userdata);
 
-/** A generic free callback */
-typedef void (*pa_free_cb_t)(void *p);
-
 /** A generic request callback */
 typedef void (*pa_stream_request_cb_t)(pa_stream *p, size_t length, void *userdata);
 
diff --git a/src/polyp/subscribe.c b/src/polyp/subscribe.c
index 110d4e52..21f5f7a5 100644
--- a/src/polyp/subscribe.c
+++ b/src/polyp/subscribe.c
@@ -75,7 +75,7 @@ pa_operation* pa_context_subscribe(pa_context *c, pa_subscription_mask_t m, pa_c
     t = pa_tagstruct_command(c, PA_COMMAND_SUBSCRIBE, &tag);
     pa_tagstruct_putu32(t, m);
     pa_pstream_send_tagstruct(c->pstream, t);
-    pa_pdispatch_register_reply(c->pdispatch, tag, DEFAULT_TIMEOUT, pa_context_simple_ack_callback, pa_operation_ref(o));
+    pa_pdispatch_register_reply(c->pdispatch, tag, DEFAULT_TIMEOUT, pa_context_simple_ack_callback, pa_operation_ref(o), (pa_free_cb_t) pa_operation_unref);
 
     return o;
 }
diff --git a/src/polypcore/pdispatch.c b/src/polypcore/pdispatch.c
index a4e58f8c..b087f1a5 100644
--- a/src/polypcore/pdispatch.c
+++ b/src/polypcore/pdispatch.c
@@ -97,6 +97,7 @@ struct reply_info {
     PA_LLIST_FIELDS(struct reply_info);
     pa_pdispatch_cb_t callback;
     void *userdata;
+    pa_free_cb_t free_cb;
     uint32_t tag;
     pa_time_event *time_event;
 };
@@ -145,8 +146,12 @@ pa_pdispatch* pa_pdispatch_new(pa_mainloop_api *mainloop, const pa_pdispatch_cb_
 static void pdispatch_free(pa_pdispatch *pd) {
     assert(pd);
 
-    while (pd->replies)
+    while (pd->replies) {
+        if (pd->replies->free_cb)
+            pd->replies->free_cb(pd->replies->userdata);
+
         reply_info_free(pd->replies);
+    }
     
     pa_xfree(pd);
 }
@@ -243,7 +248,7 @@ static void timeout_callback(pa_mainloop_api*m, pa_time_event*e, PA_GCC_UNUSED c
     run_action(r->pdispatch, r, PA_COMMAND_TIMEOUT, NULL);
 }
 
-void pa_pdispatch_register_reply(pa_pdispatch *pd, uint32_t tag, int timeout, pa_pdispatch_cb_t cb, void *userdata) {
+void pa_pdispatch_register_reply(pa_pdispatch *pd, uint32_t tag, int timeout, pa_pdispatch_cb_t cb, void *userdata, pa_free_cb_t free_cb) {
     struct reply_info *r;
     struct timeval tv;
     assert(pd && pd->ref >= 1 && cb);
@@ -252,6 +257,7 @@ void pa_pdispatch_register_reply(pa_pdispatch *pd, uint32_t tag, int timeout, pa
     r->pdispatch = pd;
     r->callback = cb;
     r->userdata = userdata;
+    r->free_cb = free_cb;
     r->tag = tag;
     
     pa_gettimeofday(&tv);
diff --git a/src/polypcore/pdispatch.h b/src/polypcore/pdispatch.h
index aa898abf..0074e8b3 100644
--- a/src/polypcore/pdispatch.h
+++ b/src/polypcore/pdispatch.h
@@ -24,6 +24,7 @@
 
 #include <inttypes.h>
 #include <polyp/mainloop-api.h>
+#include <polyp/def.h>
 #include <polypcore/tagstruct.h>
 #include <polypcore/packet.h>
 
@@ -38,7 +39,7 @@ pa_pdispatch* pa_pdispatch_ref(pa_pdispatch *pd);
 
 int pa_pdispatch_run(pa_pdispatch *pd, pa_packet*p, const void*creds, void *userdata);
 
-void pa_pdispatch_register_reply(pa_pdispatch *pd, uint32_t tag, int timeout, pa_pdispatch_cb_t callback, void *userdata);
+void pa_pdispatch_register_reply(pa_pdispatch *pd, uint32_t tag, int timeout, pa_pdispatch_cb_t callback, void *userdata, pa_free_cb_t free_cb);
 
 int pa_pdispatch_is_pending(pa_pdispatch *pd);