proper ref counting for more objects

some documentation update


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

doxygen/doxygen.conf.in

@@ -291,7 +291,7 @@ HIDE_SCOPE_NAMES       = NO
 # will put a list of the files that are included by a file in the documentation 
 # of that file.
 
-SHOW_INCLUDE_FILES     = YES
+SHOW_INCLUDE_FILES     = NO
 
 # If the INLINE_INFO tag is set to YES (the default) then a tag [inline] 
 # is inserted in the documentation for inline members.

polyp/Makefile.am

@@ -124,7 +124,8 @@ polypaudio_SOURCES = idxset.c idxset.h \
 		play-memchunk.c play-memchunk.h \
 		autoload.c autoload.h \
 		xmalloc.c xmalloc.h \
-		subscribe.h subscribe.c
+		subscribe.h subscribe.c \
+		debug.h
 
 polypaudio_CFLAGS = $(AM_CFLAGS) $(LIBSAMPLERATE_CFLAGS) $(LIBSNDFILE_CFLAGS)
 polypaudio_INCLUDES = $(INCLTDL)
@@ -296,7 +297,8 @@ libpolyp_la_SOURCES = polyplib.h \
 		polyplib-introspect.c polyplib-introspect.h \
 		polyplib-scache.c polyplib-scache.h \
 		polyplib-subscribe.c polyplib-subscribe.h \
-		cdecl.h
+		cdecl.h \
+		llist.h
 libpolyp_la_CFLAGS = $(AM_CFLAGS)
 
 libpolyp_mainloop_la_SOURCES = mainloop-api.h mainloop-api.c \

polyp/client.c

@@ -79,4 +79,6 @@ void pa_client_rename(struct pa_client *c, const char *name) {
     assert(c);
     pa_xfree(c->name);
     c->name = pa_xstrdup(name);
+
+    pa_subscription_post(c->core, PA_SUBSCRIPTION_EVENT_CLIENT|PA_SUBSCRIPTION_EVENT_CHANGE, c->index);
 }

polyp/glib-mainloop.h

@@ -24,6 +24,6 @@ void pa_glib_mainloop_free(struct pa_glib_mainloop* g);
 /** Return the abstract main loop API vtable for the GLIB main loop object */
 struct pa_mainloop_api* pa_glib_mainloop_get_api(struct pa_glib_mainloop *g);
 
-PA_C_DECL_BEGIN
+PA_C_DECL_END
 
 #endif

polyp/mainloop-api.h

@@ -27,6 +27,20 @@
 
 #include "cdecl.h"
 
+/** \file
+ * 
+ * Main loop abstraction layer. Both the polypaudio core and the
+ * polypaudio client library use a main loop abstraction layer. Due to
+ * this it is possible to embed polypaudio into other
+ * applications easily. Two main loop implemenations are
+ * currently available:
+ * \li A minimal implementation based on the C library's poll() function (See \ref mainloop.h)
+ * \li A wrapper around the GLIB main loop. Use this to embed polypaudio into your GLIB/GTK+/GNOME programs (See \ref glib-mainloop.h)
+ *
+ * The structure pa_mainloop_api is used as vtable for the main loop abstraction.
+ * */
+
+
 PA_C_DECL_BEGIN
 
 /** A bitmask for IO events */
@@ -55,16 +69,28 @@ struct pa_mainloop_api {
     /** A pointer to some private, arbitrary data of the main loop implementation */
     void *userdata;
 
-    /* IO sources */
+    /** Create a new IO event source object */
     struct pa_io_event* (*io_new)(struct pa_mainloop_api*a, int fd, enum pa_io_event_flags events, void (*callback) (struct pa_mainloop_api*a, struct pa_io_event* e, int fd, enum pa_io_event_flags events, void *userdata), void *userdata);
+
+    /** Enable or disable IO events on this object */
     void (*io_enable)(struct pa_io_event* e, enum pa_io_event_flags events);
+
+    /** Free a IO event source object */
     void (*io_free)(struct pa_io_event* e);
+
+    /** Set a function that is called when the IO event source is destroyed. Use this to free the userdata argument if required */
     void (*io_set_destroy)(struct pa_io_event *e, void (*callback) (struct pa_mainloop_api*a, struct pa_io_event *e, void *userdata));
 
-    /* Time sources */
+    /** Create a new timer event source object for the specified Unix time */
     struct pa_time_event* (*time_new)(struct pa_mainloop_api*a, const struct timeval *tv, void (*callback) (struct pa_mainloop_api*a, struct pa_time_event* e, const struct timeval *tv, void *userdata), void *userdata);
+
+    /** Restart a running or expired timer event source with a new Unix time */
     void (*time_restart)(struct pa_time_event* e, const struct timeval *tv);
+
+    /** Free a deferred timer event source object */
     void (*time_free)(struct pa_time_event* e);
+
+    /** Set a function that is called when the timer event source is destroyed. Use this to free the userdata argument if required */
     void (*time_set_destroy)(struct pa_time_event *e, void (*callback) (struct pa_mainloop_api*a, struct pa_time_event *e, void *userdata));
 
     /** Create a new deferred event source object */

polyp/mainloop-signal.h

@@ -23,6 +23,9 @@
 ***/
 
 #include "mainloop-api.h"
+#include "cdecl.h"
+
+PA_C_DECL_BEGIN
 
 int pa_signal_init(struct pa_mainloop_api *api);
 void pa_signal_done(void);
@@ -34,4 +37,6 @@ void pa_signal_free(struct pa_signal_event *e);
 
 void pa_signal_set_destroy(struct pa_signal_event *e, void (*callback) (struct pa_mainloop_api *api, struct pa_signal_event*e, void *userdata));
 
+PA_C_DECL_END
+
 #endif

polyp/module-protocol-stub.c

@@ -145,7 +145,7 @@ int pa_module_init(struct pa_core *c, struct pa_module*m) {
         goto finish;
 
     if (!(m->userdata = protocol_new(c, s, m, ma))) {
-        pa_socket_server_free(s);
+        pa_socket_server_unref(s);
         goto finish;
     }
 

polyp/module-x11-bell.c

@@ -154,6 +154,7 @@ void pa_module_done(struct pa_core *c, struct pa_module*m) {
     }
 
     pa_xfree(u->scache_item);
+    pa_xfree(u->sink_name);
     
     if (u->display)
         XCloseDisplay(u->display);

polyp/native-common.h

@@ -22,6 +22,10 @@
   USA.
 ***/
 
+#include "cdecl.h"
+
+PA_C_DECL_BEGIN
+
 enum {
     PA_COMMAND_ERROR,
     PA_COMMAND_TIMEOUT, /* pseudo command */
@@ -121,4 +125,6 @@ enum pa_subscription_event_type {
 
 #define pa_subscription_match_flags(m, t) (!!((m) & (1 << ((t) & PA_SUBSCRIPTION_EVENT_FACILITY_MASK))))
 
+PA_C_DECL_END
+
 #endif

polyp/pdispatch.c

@@ -30,6 +30,7 @@
 #include "pdispatch.h"
 #include "native-common.h"
 #include "xmalloc.h"
+#include "llist.h"
 
 /*#define DEBUG_OPCODES*/
 
@@ -65,37 +66,30 @@ static const char *command_names[PA_COMMAND_MAX] = {
 
 struct reply_info {
     struct pa_pdispatch *pdispatch;
-    struct reply_info *next, *previous;
+    PA_LLIST_FIELDS(struct reply_info);
     void (*callback)(struct pa_pdispatch *pd, uint32_t command, uint32_t tag, struct pa_tagstruct *t, void *userdata);
     void *userdata;
     uint32_t tag;
     struct pa_time_event *time_event;
-    int callback_is_running;
 };
 
 struct pa_pdispatch {
+    int ref;
     struct pa_mainloop_api *mainloop;
     const struct pa_pdispatch_command *command_table;
     unsigned n_commands;
-    struct reply_info *replies;
+    PA_LLIST_HEAD(struct reply_info, replies);
     void (*drain_callback)(struct pa_pdispatch *pd, void *userdata);
     void *drain_userdata;
-    int in_use, shall_free;
 };
 
 static void reply_info_free(struct reply_info *r) {
     assert(r && r->pdispatch && r->pdispatch->mainloop);
 
-    if (r->pdispatch)
+    if (r->time_event)
         r->pdispatch->mainloop->time_free(r->time_event);
-
+    
-    if (r->previous)
+    PA_LLIST_REMOVE(struct reply_info, r->pdispatch->replies, r);
-        r->previous->next = r->next;
-    else
-        r->pdispatch->replies = r->next;
-
-    if (r->next)
-        r->next->previous = r->previous;
     
     pa_xfree(r);
 }
@@ -107,37 +101,56 @@ struct pa_pdispatch* pa_pdispatch_new(struct pa_mainloop_api *mainloop, const st
     assert((entries && table) || (!entries && !table));
     
     pd = pa_xmalloc(sizeof(struct pa_pdispatch));
+    pd->ref = 1;
     pd->mainloop = mainloop;
     pd->command_table = table;
     pd->n_commands = entries;
-    pd->replies = NULL;
+    PA_LLIST_HEAD_INIT(struct pa_reply_info, pd->replies);
     pd->drain_callback = NULL;
     pd->drain_userdata = NULL;
 
-    pd->in_use = pd->shall_free = 0;
-
     return pd;
 }
 
-void pa_pdispatch_free(struct pa_pdispatch *pd) {
+void pdispatch_free(struct pa_pdispatch *pd) {
     assert(pd);
 
-    if (pd->in_use) {
-        pd->shall_free = 1;
-        return;
-    }
-    
     while (pd->replies)
         reply_info_free(pd->replies);
+    
     pa_xfree(pd);
 }
 
+static void run_action(struct pa_pdispatch *pd, struct reply_info *r, uint32_t command, struct pa_tagstruct *ts) {
+    void (*callback)(struct pa_pdispatch *pd, uint32_t command, uint32_t tag, struct pa_tagstruct *t, void *userdata);
+    void *userdata;
+    uint32_t tag;
+    assert(r);
+
+    pa_pdispatch_ref(pd);
+    
+    callback = r->callback;
+    userdata = r->userdata;
+    tag = r->tag;
+    
+    reply_info_free(r);
+    
+    callback(pd, command, tag, ts, userdata);
+
+    if (pd->drain_callback && !pa_pdispatch_is_pending(pd))
+        pd->drain_callback(pd, pd->drain_userdata);
+
+    pa_pdispatch_unref(pd);
+}
+
 int pa_pdispatch_run(struct pa_pdispatch *pd, struct pa_packet*packet, void *userdata) {
     uint32_t tag, command;
     struct pa_tagstruct *ts = NULL;
     int ret = -1;
-    assert(pd && packet && packet->data && !pd->in_use);
+    assert(pd && packet && packet->data);
 
+    pa_pdispatch_ref(pd);
+    
     if (packet->length <= 8)
         goto finish;
 
@@ -159,20 +172,8 @@ int pa_pdispatch_run(struct pa_pdispatch *pd, struct pa_packet*packet, void *use
             if (r->tag == tag)
                 break;
 
-        if (r) {
+        if (r)
-            pd->in_use = r->callback_is_running = 1;
+            run_action(pd, r, command, ts);
-            assert(r->callback);
-            r->callback(r->pdispatch, command, tag, ts, r->userdata);
-            pd->in_use = r->callback_is_running = 0;
-            reply_info_free(r);
-            
-            if (pd->shall_free)
-                pa_pdispatch_free(pd);
-            else {
-                if (pd->drain_callback && !pa_pdispatch_is_pending(pd))
-                    pd->drain_callback(pd, pd->drain_userdata);
-            }
-        }
 
     } else if (pd->command_table && command < pd->n_commands) {
         const struct pa_pdispatch_command *c = pd->command_table+command;
@@ -186,33 +187,30 @@ int pa_pdispatch_run(struct pa_pdispatch *pd, struct pa_packet*packet, void *use
         
 finish:
     if (ts)
-        pa_tagstruct_free(ts);    
+        pa_tagstruct_free(ts);
+
+    pa_pdispatch_unref(pd);
 
     return ret;
 }
 
 static void timeout_callback(struct pa_mainloop_api*m, struct pa_time_event*e, const struct timeval *tv, void *userdata) {
     struct reply_info*r = userdata;
-    assert (r && r->time_event == e && r->pdispatch && r->pdispatch->mainloop == m && r->callback);
+    assert(r && r->time_event == e && r->pdispatch && r->pdispatch->mainloop == m && r->callback);
 
-    r->callback(r->pdispatch, PA_COMMAND_TIMEOUT, r->tag, NULL, r->userdata);
+    run_action(r->pdispatch, r, PA_COMMAND_TIMEOUT, NULL);
-    reply_info_free(r);
-
-    if (r->pdispatch->drain_callback && !pa_pdispatch_is_pending(r->pdispatch))
-        r->pdispatch->drain_callback(r->pdispatch, r->pdispatch->drain_userdata);
 }
 
 void pa_pdispatch_register_reply(struct pa_pdispatch *pd, uint32_t tag, int timeout, void (*cb)(struct pa_pdispatch *pd, uint32_t command, uint32_t tag, struct pa_tagstruct *t, void *userdata), void *userdata) {
     struct reply_info *r;
     struct timeval tv;
-    assert(pd && cb);
+    assert(pd && pd->ref >= 1 && cb);
 
     r = pa_xmalloc(sizeof(struct reply_info));
     r->pdispatch = pd;
     r->callback = cb;
     r->userdata = userdata;
     r->tag = tag;
-    r->callback_is_running = 0;
     
     gettimeofday(&tv, NULL);
     tv.tv_sec += timeout;
@@ -220,11 +218,7 @@ void pa_pdispatch_register_reply(struct pa_pdispatch *pd, uint32_t tag, int time
     r->time_event = pd->mainloop->time_new(pd->mainloop, &tv, timeout_callback, r);
     assert(r->time_event);
 
-    r->previous = NULL;
+    PA_LLIST_PREPEND(struct reply_info, pd->replies, r);
-    r->next = pd->replies;
-    if (r->next)
-        r->next->previous = r;
-    pd->replies = r;
 }
 
 int pa_pdispatch_is_pending(struct pa_pdispatch *pd) {
@@ -248,7 +242,20 @@ void pa_pdispatch_unregister_reply(struct pa_pdispatch *pd, void *userdata) {
     for (r = pd->replies; r; r = n) {
         n = r->next;
 
-        if (!r->callback_is_running && r->userdata == userdata) /* when this item's callback is currently running it is destroyed anyway in the very near future */
+        if (r->userdata == userdata) 
             reply_info_free(r);
     }
 }
+
+void pa_pdispatch_unref(struct pa_pdispatch *pd) {
+    assert(pd && pd->ref >= 1);
+
+    if (!(--(pd->ref)))
+        pdispatch_free(pd);
+}
+
+struct pa_pdispatch* pa_pdispatch_ref(struct pa_pdispatch *pd) {
+    assert(pd && pd->ref >= 1);
+    pd->ref++;
+    return pd;
+}

polyp/pdispatch.h

@@ -29,14 +29,13 @@
 
 struct pa_pdispatch;
 
-/* It is safe to destroy the calling pdispatch object from all callbacks */
-
 struct pa_pdispatch_command {
     void (*proc)(struct pa_pdispatch *pd, uint32_t command, uint32_t tag, struct pa_tagstruct *t, void *userdata);
 };
 
 struct pa_pdispatch* pa_pdispatch_new(struct pa_mainloop_api *m, const struct pa_pdispatch_command*table, unsigned entries);
-void pa_pdispatch_free(struct pa_pdispatch *pd);
+void pa_pdispatch_unref(struct pa_pdispatch *pd);
+struct pa_pdispatch* pa_pdispatch_ref(struct pa_pdispatch *pd);
 
 int pa_pdispatch_run(struct pa_pdispatch *pd, struct pa_packet*p, void *userdata);
 

polyp/polyplib-context.c

@@ -92,11 +92,13 @@ static void context_free(struct pa_context *c) {
         pa_stream_set_state(c->streams, PA_STREAM_TERMINATED);
     
     if (c->client)
-        pa_socket_client_free(c->client);
+        pa_socket_client_unref(c->client);
     if (c->pdispatch)
-        pa_pdispatch_free(c->pdispatch);
+        pa_pdispatch_unref(c->pdispatch);
-    if (c->pstream)
+    if (c->pstream) {
-        pa_pstream_free(c->pstream);
+        pa_pstream_close(c->pstream);
+        pa_pstream_unref(c->pstream);
+    }
     
     if (c->record_streams)
         pa_dynarray_free(c->record_streams, NULL, NULL);
@@ -140,15 +142,17 @@ void pa_context_set_state(struct pa_context *c, enum pa_context_state st) {
         }
 
         if (c->pdispatch)
-            pa_pdispatch_free(c->pdispatch);
+            pa_pdispatch_unref(c->pdispatch);
         c->pdispatch = NULL;
     
-        if (c->pstream)
+        if (c->pstream) {
-            pa_pstream_free(c->pstream);
+            pa_pstream_close(c->pstream);
+            pa_pstream_unref(c->pstream);
+        }
         c->pstream = NULL;
     
         if (c->client)
-            pa_socket_client_free(c->client);
+            pa_socket_client_unref(c->client);
         c->client = NULL;
     }
 
@@ -267,7 +271,7 @@ static void on_connection(struct pa_socket_client *client, struct pa_iochannel*i
 
     pa_context_ref(c);
     
-    pa_socket_client_free(client);
+    pa_socket_client_unref(client);
     c->client = NULL;
 
     if (!io) {

polyp/polyplib-error.h

@@ -25,11 +25,11 @@
 #include <inttypes.h>
 #include "cdecl.h"
 
-PA_C_DECL_BEGIN;
+PA_C_DECL_BEGIN
 
 /** Return a human readable error message for the specified numeric error code */
 const char* pa_strerror(uint32_t error);
 
-PA_C_DECL_END;
+PA_C_DECL_END
 
 #endif

polyp/polyplib-subscribe.c

@@ -24,6 +24,7 @@
 #endif
 
 #include <assert.h>
+#include <stdio.h>
 
 #include "polyplib-subscribe.h"
 #include "polyplib-internal.h"
@@ -65,10 +66,15 @@ struct pa_operation* pa_context_subscribe(struct pa_context *c, enum pa_subscrip
     t = pa_tagstruct_new(NULL, 0);
     pa_tagstruct_putu32(t, PA_COMMAND_SUBSCRIBE);
     pa_tagstruct_putu32(t, tag = c->ctag++);
-    pa_tagstruct_putu32(t, cb ? m : 0);
+    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, o);
 
     return pa_operation_ref(o);
 }
 
+void pa_context_set_subscribe_callback(struct pa_context *c, void (*cb)(struct pa_context *c, enum pa_subscription_event_type t, uint32_t index, void *userdata), void *userdata) {
+    assert(c);
+    c->subscribe_callback = cb;
+    c->subscribe_userdata = userdata;
+}

polyp/protocol-cli.c

@@ -80,6 +80,6 @@ void pa_protocol_cli_free(struct pa_protocol_cli *p) {
     assert(p);
 
     pa_idxset_free(p->connections, free_connection, NULL);
-    pa_socket_server_free(p->server);
+    pa_socket_server_unref(p->server);
     pa_xfree(p);
 }

polyp/protocol-esound.c

@@ -1024,6 +1024,6 @@ void pa_protocol_esound_free(struct pa_protocol_esound *p) {
         connection_free(c);
 
     pa_idxset_free(p->connections, NULL, NULL);
-    pa_socket_server_free(p->server);
+    pa_socket_server_unref(p->server);
     pa_xfree(p);
 }

polyp/protocol-native.c

@@ -306,8 +306,9 @@ static void connection_free(struct connection *c) {
             upload_stream_free((struct upload_stream*) o);
     pa_idxset_free(c->output_streams, NULL, NULL);
 
-    pa_pdispatch_free(c->pdispatch);
+    pa_pdispatch_unref(c->pdispatch);
-    pa_pstream_free(c->pstream);
+    pa_pstream_close(c->pstream);
+    pa_pstream_unref(c->pstream);
     pa_client_free(c->client);
 
     if (c->subscription)
@@ -1330,6 +1331,6 @@ void pa_protocol_native_free(struct pa_protocol_native *p) {
     while ((c = pa_idxset_first(p->connections, NULL)))
         connection_free(c);
     pa_idxset_free(p->connections, NULL, NULL);
-    pa_socket_server_free(p->server);
+    pa_socket_server_unref(p->server);
     pa_xfree(p);
 }

polyp/protocol-simple.c

@@ -426,7 +426,7 @@ void pa_protocol_simple_free(struct pa_protocol_simple *p) {
     }
 
     if (p->server)
-        pa_socket_server_free(p->server);
+        pa_socket_server_unref(p->server);
     pa_xfree(p);
 }
 

polyp/pstream.c

@@ -57,13 +57,13 @@ struct item_info {
 };
 
 struct pa_pstream {
+    int ref;
+    
     struct pa_mainloop_api *mainloop;
     struct pa_defer_event *defer_event;
     struct pa_iochannel *io;
     struct pa_queue *send_queue;
 
-    int in_use, shall_free;
-    
     int dead;
     void (*die_callback) (struct pa_pstream *p, void *userdata);
     void *die_callback_userdata;
@@ -97,40 +97,24 @@ static void do_write(struct pa_pstream *p);
 static void do_read(struct pa_pstream *p);
 
 static void do_something(struct pa_pstream *p) {
-    assert(p && !p->shall_free);
+    assert(p);
     p->mainloop->defer_enable(p->defer_event, 0);
 
-    if (p->dead)
+    pa_pstream_ref(p);
-        return;
+    
-
+    if (!p->dead && pa_iochannel_is_hungup(p->io)) {
-    if (pa_iochannel_is_hungup(p->io)) {
         p->dead = 1;
         if (p->die_callback)
             p->die_callback(p, p->die_callback_userdata);
-
-        return;
     }
 
-    if (pa_iochannel_is_writable(p->io)) {
+    if (!p->dead && pa_iochannel_is_writable(p->io))
-        p->in_use = 1;
         do_write(p);
-        p->in_use = 0;
 
-        if (p->shall_free) {
+    if (!p->dead && pa_iochannel_is_readable(p->io))
-            pa_pstream_free(p);
-            return;
-        }
-    }
-
-    if (pa_iochannel_is_readable(p->io)) {
-        p->in_use = 1;
         do_read(p);
-        p->in_use = 0;
+
-        if (p->shall_free) {
+    pa_pstream_unref(p);
-            pa_pstream_free(p);
-            return;
-        }
-    }
 }
 
 static void io_callback(struct pa_iochannel*io, void *userdata) {
@@ -150,7 +134,7 @@ struct pa_pstream *pa_pstream_new(struct pa_mainloop_api *m, struct pa_iochannel
     assert(io);
 
     p = pa_xmalloc(sizeof(struct pa_pstream));
-
+    p->ref = 1;
     p->io = io;
     pa_iochannel_set_callback(io, io_callback, p);
 
@@ -181,8 +165,6 @@ struct pa_pstream *pa_pstream_new(struct pa_mainloop_api *m, struct pa_iochannel
     p->drain_callback = NULL;
     p->drain_userdata = NULL;
 
-    p->in_use = p->shall_free = 0;
-
     return p;
 }
 
@@ -202,16 +184,11 @@ static void item_free(void *item, void *p) {
     pa_xfree(i);
 }
 
-void pa_pstream_free(struct pa_pstream *p) {
+static void pstream_free(struct pa_pstream *p) {
     assert(p);
 
-    if (p->in_use) {
+    pa_pstream_close(p);
-        /* If this pstream object is used by someone else on the call stack, we have to postpone the freeing */
+    
-        p->dead = p->shall_free = 1;
-        return;
-    }
-
-    pa_iochannel_free(p->io);
     pa_queue_free(p->send_queue, item_free, NULL);
 
     if (p->write.current)
@@ -223,7 +200,6 @@ void pa_pstream_free(struct pa_pstream *p) {
     if (p->read.packet)
         pa_packet_unref(p->read.packet);
 
-    p->mainloop->defer_free(p->defer_event);
     pa_xfree(p);
 }
 
@@ -456,3 +432,36 @@ void pa_pstream_set_drain_callback(struct pa_pstream *p, void (*cb)(struct pa_ps
     p->drain_userdata = userdata;
 }
 
+void pa_pstream_unref(struct pa_pstream*p) {
+    assert(p && p->ref >= 1);
+
+    if (!(--(p->ref)))
+        pstream_free(p);
+}
+
+struct pa_pstream* pa_pstream_ref(struct pa_pstream*p) {
+    assert(p && p->ref >= 1);
+    p->ref++;
+    return p;
+}
+
+void pa_pstream_close(struct pa_pstream *p) {
+    assert(p);
+
+    p->dead = 1;
+
+    if (p->io) {
+        pa_iochannel_free(p->io);
+        p->io = NULL;
+    }
+
+    if (p->defer_event) {
+        p->mainloop->defer_free(p->defer_event);
+        p->defer_event = NULL;
+    }
+
+    p->die_callback = NULL;
+    p->drain_callback = NULL;
+    p->recieve_packet_callback = NULL;
+    p->recieve_memblock_callback = NULL;
+}

polyp/pstream.h

@@ -30,12 +30,11 @@
 #include "mainloop-api.h"
 #include "memchunk.h"
 
-/* It is safe to destroy the calling pstream object from all callbacks */
-
 struct pa_pstream;
 
 struct pa_pstream* pa_pstream_new(struct pa_mainloop_api *m, struct pa_iochannel *io);
-void pa_pstream_free(struct pa_pstream*p);
+void pa_pstream_unref(struct pa_pstream*p);
+struct pa_pstream* pa_pstream_ref(struct pa_pstream*p);
 
 void pa_pstream_send_packet(struct pa_pstream*p, struct pa_packet *packet);
 void pa_pstream_send_memblock(struct pa_pstream*p, uint32_t channel, int32_t delta, const struct pa_memchunk *chunk);
@@ -48,4 +47,6 @@ void pa_pstream_set_die_callback(struct pa_pstream *p, void (*callback)(struct p
 
 int pa_pstream_is_pending(struct pa_pstream *p);
 
+void pa_pstream_close(struct pa_pstream *p);
+
 #endif

polyp/socket-client.c

@@ -39,6 +39,7 @@
 #include "xmalloc.h"
 
 struct pa_socket_client {
+    int ref;
     struct pa_mainloop_api *mainloop;
     int fd;
     struct pa_io_event *io_event;
@@ -52,6 +53,7 @@ static struct pa_socket_client*pa_socket_client_new(struct pa_mainloop_api *m) {
     assert(m);
 
     c = pa_xmalloc(sizeof(struct pa_socket_client));
+    c->ref = 1;
     c->mainloop = m;
     c->fd = -1;
     c->io_event = NULL;
@@ -62,38 +64,40 @@ static struct pa_socket_client*pa_socket_client_new(struct pa_mainloop_api *m) {
 }
 
 static void do_call(struct pa_socket_client *c) {
-    struct pa_iochannel *io;
+    struct pa_iochannel *io = NULL;
     int error, lerror;
     assert(c && c->callback);
 
+    pa_socket_client_ref(c);
+    
     lerror = sizeof(error);
     if (getsockopt(c->fd, SOL_SOCKET, SO_ERROR, &error, &lerror) < 0) {
         fprintf(stderr, "getsockopt(): %s\n", strerror(errno));
-        goto failed;
+        goto finish;
     }
 
     if (lerror != sizeof(error)) {
         fprintf(stderr, "getsocktop() returned invalid size.\n");
-        goto failed;
+        goto finish;
     }
 
     if (error != 0) {
         fprintf(stderr, "connect(): %s\n", strerror(error));
-        goto failed;
+        goto finish;
     }
         
     io = pa_iochannel_new(c->mainloop, c->fd, c->fd);
     assert(io);
-    c->fd = -1;
-    c->callback(c, io, c->userdata);
-
-    return;
     
-failed:
+finish:
-    close(c->fd);
+    if (!io)
+        close(c->fd);
     c->fd = -1;
-    c->callback(c, NULL, c->userdata);
+    
-    return;
+    assert(c->callback);
+    c->callback(c, io, c->userdata);
+    
+    pa_socket_client_unref(c);
 }
 
 static void connect_fixed_cb(struct pa_mainloop_api *m, struct pa_defer_event *e, void *userdata) {
@@ -159,7 +163,7 @@ struct pa_socket_client* pa_socket_client_new_ipv4(struct pa_mainloop_api *m, ui
     return c;
 
 fail:
-    pa_socket_client_free(c);
+    pa_socket_client_unref(c);
     return NULL;
 }
 
@@ -188,7 +192,7 @@ struct pa_socket_client* pa_socket_client_new_unix(struct pa_mainloop_api *m, co
     return c;
 
 fail:
-    pa_socket_client_free(c);
+    pa_socket_client_unref(c);
     return NULL;
 }
 
@@ -214,12 +218,12 @@ struct pa_socket_client* pa_socket_client_new_sockaddr(struct pa_mainloop_api *m
     return c;
 
 fail:
-    pa_socket_client_free(c);
+    pa_socket_client_unref(c);
     return NULL;
     
 }
 
-void pa_socket_client_free(struct pa_socket_client *c) {
+void socket_client_free(struct pa_socket_client *c) {
     assert(c && c->mainloop);
     if (c->io_event)
         c->mainloop->io_free(c->io_event);
@@ -230,6 +234,19 @@ void pa_socket_client_free(struct pa_socket_client *c) {
     pa_xfree(c);
 }
 
+void pa_socket_client_unref(struct pa_socket_client *c) {
+    assert(c && c->ref >= 1);
+
+    if (!(--(c->ref)))
+        socket_client_free(c);
+}
+
+struct pa_socket_client* pa_socket_client_ref(struct pa_socket_client *c) {
+    assert(c && c->ref >= 1);
+    c->ref++;
+    return c;
+}
+
 void pa_socket_client_set_callback(struct pa_socket_client *c, void (*on_connection)(struct pa_socket_client *c, struct pa_iochannel*io, void *userdata), void *userdata) {
     assert(c);
     c->callback = on_connection;

polyp/socket-client.h

@@ -36,7 +36,8 @@ struct pa_socket_client* pa_socket_client_new_ipv4(struct pa_mainloop_api *m, ui
 struct pa_socket_client* pa_socket_client_new_unix(struct pa_mainloop_api *m, const char *filename);
 struct pa_socket_client* pa_socket_client_new_sockaddr(struct pa_mainloop_api *m, const struct sockaddr *sa, size_t salen);
 
-void pa_socket_client_free(struct pa_socket_client *c);
+void pa_socket_client_unref(struct pa_socket_client *c);
+struct pa_socket_client* pa_socket_client_ref(struct pa_socket_client *c);
 
 void pa_socket_client_set_callback(struct pa_socket_client *c, void (*on_connection)(struct pa_socket_client *c, struct pa_iochannel*io, void *userdata), void *userdata);
 

polyp/socket-server.c

@@ -40,6 +40,7 @@
 #include "xmalloc.h"
 
 struct pa_socket_server {
+    int ref;
     int fd;
     char *filename;
 
@@ -57,14 +58,16 @@ static void callback(struct pa_mainloop_api *mainloop, struct pa_io_event *e, in
     int nfd;
     assert(s && s->mainloop == mainloop && s->io_event == e && e && fd >= 0 && fd == s->fd);
 
+    pa_socket_server_ref(s);
+    
     if ((nfd = accept(fd, NULL, NULL)) < 0) {
         fprintf(stderr, "accept(): %s\n", strerror(errno));
-        return;
+        goto finish;
     }
 
     if (!s->on_connection) {
         close(nfd);
-        return;
+        goto finish;
     }
 
     /* There should be a check for socket type here */
@@ -76,6 +79,9 @@ static void callback(struct pa_mainloop_api *mainloop, struct pa_io_event *e, in
     io = pa_iochannel_new(s->mainloop, nfd, nfd);
     assert(io);
     s->on_connection(s, io, s->userdata);
+
+finish:
+    pa_socket_server_unref(s);
 }
 
 struct pa_socket_server* pa_socket_server_new(struct pa_mainloop_api *m, int fd) {
@@ -83,6 +89,7 @@ struct pa_socket_server* pa_socket_server_new(struct pa_mainloop_api *m, int fd)
     assert(m && fd >= 0);
     
     s = pa_xmalloc(sizeof(struct pa_socket_server));
+    s->ref = 1;
     s->fd = fd;
     s->filename = NULL;
     s->on_connection = NULL;
@@ -97,6 +104,12 @@ struct pa_socket_server* pa_socket_server_new(struct pa_mainloop_api *m, int fd)
     return s;
 }
 
+struct pa_socket_server* pa_socket_server_ref(struct pa_socket_server *s) {
+    assert(s && s->ref >= 1);
+    s->ref++;
+    return s;
+}
+
 struct pa_socket_server* pa_socket_server_new_unix(struct pa_mainloop_api *m, const char *filename) {
     int fd = -1;
     struct sockaddr_un sa;
@@ -184,7 +197,7 @@ fail:
     return NULL;
 }
 
-void pa_socket_server_free(struct pa_socket_server*s) {
+static void socket_server_free(struct pa_socket_server*s) {
     assert(s);
     close(s->fd);
 
@@ -197,8 +210,15 @@ void pa_socket_server_free(struct pa_socket_server*s) {
     pa_xfree(s);
 }
 
+void pa_socket_server_unref(struct pa_socket_server *s) {
+    assert(s && s->ref >= 1);
+
+    if (!(--(s->ref)))
+        socket_server_free(s);
+}
+
 void pa_socket_server_set_callback(struct pa_socket_server*s, void (*on_connection)(struct pa_socket_server*s, struct pa_iochannel *io, void *userdata), void *userdata) {
-    assert(s);
+    assert(s && s->ref >= 1);
 
     s->on_connection = on_connection;
     s->userdata = userdata;

polyp/socket-server.h

@@ -34,7 +34,8 @@ struct pa_socket_server* pa_socket_server_new(struct pa_mainloop_api *m, int fd)
 struct pa_socket_server* pa_socket_server_new_unix(struct pa_mainloop_api *m, const char *filename);
 struct pa_socket_server* pa_socket_server_new_ipv4(struct pa_mainloop_api *m, uint32_t address, uint16_t port);
 
-void pa_socket_server_free(struct pa_socket_server*s);
+void pa_socket_server_unref(struct pa_socket_server*s);
+struct pa_socket_server* pa_socket_server_ref(struct pa_socket_server *s);
 
 void pa_socket_server_set_callback(struct pa_socket_server*s, void (*on_connection)(struct pa_socket_server*s, struct pa_iochannel *io, void *userdata), void *userdata);