module-tunnel: restart module

Defer the creation of the source/sink until after the TCP connection has
been established.  Upon protocol errors, try restarting the module.

Part-of: <https://gitlab.freedesktop.org/pulseaudio/pulseaudio/-/merge_requests/688>

2 files changed, 360 insertions, 183 deletions

diff --git a/src/modules/meson.build b/src/modules/meson.build
index 0ccbef070..cdf372c8e 100644
--- a/src/modules/meson.build
+++ b/src/modules/meson.build
@@ -54,9 +54,9 @@ all_modules = [
   [ 'module-suspend-on-idle', 'module-suspend-on-idle.c' ],
   [ 'module-switch-on-connect', 'module-switch-on-connect.c' ],
   [ 'module-switch-on-port-available', 'module-switch-on-port-available.c' ],
-  [ 'module-tunnel-sink', 'module-tunnel.c', [], ['-DTUNNEL_SINK=1'], [x11_dep] ],
+  [ 'module-tunnel-sink', ['module-tunnel.c', 'restart-module.c'], [], ['-DTUNNEL_SINK=1'], [x11_dep] ],
   [ 'module-tunnel-sink-new', ['module-tunnel-sink-new.c', 'restart-module.c'] ],
-  [ 'module-tunnel-source', 'module-tunnel.c', [], [], [x11_dep] ],
+  [ 'module-tunnel-source', ['module-tunnel.c', 'restart-module.c'], [], [], [x11_dep] ],
   [ 'module-tunnel-source-new', ['module-tunnel-source-new.c', 'restart-module.c'] ],
   [ 'module-virtual-sink', 'module-virtual-sink.c' ],
   [ 'module-virtual-source', 'module-virtual-source.c' ],
diff --git a/src/modules/module-tunnel.c b/src/modules/module-tunnel.c
index e9d01ef88..0e697bd4a 100644
--- a/src/modules/module-tunnel.c
+++ b/src/modules/module-tunnel.c
@@ -22,6 +22,8 @@
 #include <config.h>
 #endif
 
+#include "restart-module.h"
+
 #include <unistd.h>
 #include <string.h>
 #include <errno.h>
@@ -81,6 +83,7 @@ PA_MODULE_USAGE(
         "auto=<determine server/sink/cookie automatically> "
         "server=<address> "
         "sink=<remote sink name> "
+        "reconnect_interval_ms=<interval to try reconnects, 0 or omitted if disabled> "
         "cookie=<filename> "
         "format=<sample format> "
         "channels=<number of channels> "
@@ -95,6 +98,7 @@ PA_MODULE_USAGE(
         "auto=<determine server/source/cookie automatically> "
         "server=<address> "
         "source=<remote source name> "
+        "reconnect_interval_ms=<interval to try reconnects, 0 or omitted if disabled> "
         "cookie=<filename> "
         "format=<sample format> "
         "channels=<number of channels> "
@@ -115,6 +119,7 @@ static const char* const valid_modargs[] = {
     "channels",
     "rate",
     "latency_msec",
+    "reconnect_interval_ms",
 #ifdef TUNNEL_SINK
     "sink_name",
     "sink_properties",
@@ -141,7 +146,8 @@ enum {
     SINK_MESSAGE_REMOTE_SUSPEND,
     SINK_MESSAGE_UPDATE_LATENCY,
     SINK_MESSAGE_GET_LATENCY_SNAPSHOT,
-    SINK_MESSAGE_POST
+    SINK_MESSAGE_POST,
+    SINK_MESSAGE_CREATED
 };
 
 #define DEFAULT_LATENCY_MSEC 100
@@ -152,13 +158,33 @@ enum {
     SOURCE_MESSAGE_POST = PA_SOURCE_MESSAGE_MAX,
     SOURCE_MESSAGE_REMOTE_SUSPEND,
     SOURCE_MESSAGE_UPDATE_LATENCY,
-    SOURCE_MESSAGE_GET_LATENCY_SNAPSHOT
+    SOURCE_MESSAGE_GET_LATENCY_SNAPSHOT,
+    SOURCE_MESSAGE_CREATED
 };
 
 #define DEFAULT_LATENCY_MSEC 25
 
 #endif
 
+struct tunnel_msg {
+    pa_msgobject parent;
+};
+
+typedef struct tunnel_msg tunnel_msg;
+PA_DEFINE_PRIVATE_CLASS(tunnel_msg, pa_msgobject);
+
+enum {
+#ifdef TUNNEL_SINK
+    TUNNEL_MESSAGE_CREATE_SINK_REQUEST,
+#else
+    TUNNEL_MESSAGE_CREATE_SOURCE_REQUEST,
+#endif
+    TUNNEL_MESSAGE_MAYBE_RESTART,
+};
+
+static int do_init(pa_module *m);
+static void do_done(pa_module *m);
+
 #ifdef TUNNEL_SINK
 static void command_request(pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata);
 static void command_started(pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata);
@@ -207,10 +233,12 @@ struct userdata {
     char *server_name;
 #ifdef TUNNEL_SINK
     char *sink_name;
+    char *configured_sink_name;
     pa_sink *sink;
     size_t requested_bytes;
 #else
     char *source_name;
+    char *configured_source_name;
     pa_source *source;
     pa_mcalign *mcalign;
 #endif
@@ -229,6 +257,7 @@ struct userdata {
 
     bool remote_corked:1;
     bool remote_suspended:1;
+    bool shutting_down:1;
 
     pa_usec_t transport_usec; /* maintained in the main thread */
     pa_usec_t thread_transport_usec; /* maintained in the IO thread */
@@ -252,12 +281,42 @@ struct userdata {
     uint32_t tlength;
     uint32_t minreq;
     uint32_t prebuf;
+
+    pa_proplist *sink_proplist;
 #else
     uint32_t fragsize;
+
+    pa_proplist *source_proplist;
 #endif
+
+    pa_sample_spec sample_spec;
+    pa_channel_map channel_map;
+
+    tunnel_msg *msg;
+
+    pa_iochannel *io;
+
+    pa_usec_t reconnect_interval_us;
 };
 
 static void request_latency(struct userdata *u);
+#ifdef TUNNEL_SINK
+static void on_sink_created(struct userdata *u);
+#else
+static void on_source_created(struct userdata *u);
+#endif
+
+/* Do a reinit of the module.  Note that u will be freed as a result of this
+ * call, while pu will live on to the next iteration.  It's up to do_done to
+ * copy anything that we want to persist across iterations out of u and into pu
+ */
+static void unload_module(struct userdata *u) {
+    if (u->reconnect_interval_us > 0) {
+        pa_restart_module_reinit(u->module, do_init, do_done, u->reconnect_interval_us);
+    } else {
+        pa_module_unload_request(u->module, true);
+    }
+}
 
 /* Called from main context */
 static void command_stream_or_client_event(pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata) {
@@ -274,7 +333,7 @@ static void command_stream_killed(pa_pdispatch *pd,  uint32_t command,  uint32_t
     pa_assert(u->pdispatch == pd);
 
     pa_log_warn("Stream killed");
-    pa_module_unload_request(u->module, true);
+    unload_module(u);
 }
 
 /* Called from main context */
@@ -306,7 +365,7 @@ static void command_suspended(pa_pdispatch *pd,  uint32_t command,  uint32_t tag
         !pa_tagstruct_eof(t)) {
 
         pa_log("Invalid packet.");
-        pa_module_unload_request(u->module, true);
+        unload_module(u);
         return;
     }
 
@@ -339,7 +398,7 @@ static void command_moved(pa_pdispatch *pd,  uint32_t command,  uint32_t tag, pa
         pa_tagstruct_get_boolean(t, &suspended) < 0) {
 
         pa_log_error("Invalid packet.");
-        pa_module_unload_request(u->module, true);
+        unload_module(u);
         return;
     }
 
@@ -368,7 +427,7 @@ static void command_stream_buffer_attr_changed(pa_pdispatch *pd, uint32_t comman
         pa_tagstruct_getu32(t, &maxlength) < 0) {
 
         pa_log_error("Invalid packet.");
-        pa_module_unload_request(u->module, true);
+        unload_module(u);
         return;
     }
 
@@ -377,7 +436,7 @@ static void command_stream_buffer_attr_changed(pa_pdispatch *pd, uint32_t comman
             pa_tagstruct_get_usec(t, &usec) < 0) {
 
             pa_log_error("Invalid packet.");
-            pa_module_unload_request(u->module, true);
+            unload_module(u);
             return;
         }
     } else {
@@ -387,7 +446,7 @@ static void command_stream_buffer_attr_changed(pa_pdispatch *pd, uint32_t comman
             pa_tagstruct_get_usec(t, &usec) < 0) {
 
             pa_log_error("Invalid packet.");
-            pa_module_unload_request(u->module, true);
+            unload_module(u);
             return;
         }
     }
@@ -613,6 +672,12 @@ static int sink_process_msg(pa_msgobject *o, int code, void *data, int64_t offse
             u->receive_counter += chunk->length;
 
             return 0;
+
+        case SINK_MESSAGE_CREATED:
+
+            on_sink_created(u);
+
+            return 0;
     }
 
     return pa_sink_process_msg(o, code, data, offset, chunk);
@@ -740,6 +805,12 @@ static int source_process_msg(pa_msgobject *o, int code, void *data, int64_t off
 
             return 0;
         }
+
+        case SOURCE_MESSAGE_CREATED:
+
+            on_source_created(u);
+
+            return 0;
     }
 
     return pa_source_process_msg(o, code, data, offset, chunk);
@@ -793,7 +864,7 @@ static void thread_func(void *userdata) {
         int ret;
 
 #ifdef TUNNEL_SINK
-        if (PA_UNLIKELY(u->sink->thread_info.rewind_requested))
+        if (u->sink && PA_UNLIKELY(u->sink->thread_info.rewind_requested))
             pa_sink_process_rewind(u->sink, 0);
 #endif
 
@@ -807,7 +878,7 @@ static void thread_func(void *userdata) {
 fail:
     /* If this was no regular exit from the loop we have to continue
      * processing messages until we received PA_MESSAGE_SHUTDOWN */
-    pa_asyncmsgq_post(u->thread_mq.outq, PA_MSGOBJECT(u->core), PA_CORE_MESSAGE_UNLOAD_MODULE, u->module, 0, NULL, NULL);
+    pa_asyncmsgq_post(u->thread_mq.outq, PA_MSGOBJECT(u->msg), TUNNEL_MESSAGE_MAYBE_RESTART, u, 0, NULL, NULL);
     pa_asyncmsgq_wait_for(u->thread_mq.inq, PA_MESSAGE_SHUTDOWN);
 
 finish:
@@ -841,7 +912,7 @@ static void command_request(pa_pdispatch *pd, uint32_t command,  uint32_t tag, p
     return;
 
 fail:
-    pa_module_unload_request(u->module, true);
+    unload_module(u);
 }
 
 #endif
@@ -956,7 +1027,7 @@ static void stream_get_latency_callback(pa_pdispatch *pd, uint32_t command, uint
 
 fail:
 
-    pa_module_unload_request(u->module, true);
+    unload_module(u);
 }
 
 /* Called from main context */
@@ -1091,7 +1162,7 @@ static void server_info_cb(pa_pdispatch *pd, uint32_t command,  uint32_t tag, pa
     return;
 
 fail:
-    pa_module_unload_request(u->module, true);
+    unload_module(u);
 }
 
 static int read_ports(struct userdata *u, pa_tagstruct *t) {
@@ -1246,7 +1317,7 @@ static void sink_info_cb(pa_pdispatch *pd, uint32_t command,  uint32_t tag, pa_t
     return;
 
 fail:
-    pa_module_unload_request(u->module, true);
+    unload_module(u);
 }
 
 /* Called from main context */
@@ -1355,7 +1426,7 @@ static void sink_input_info_cb(pa_pdispatch *pd, uint32_t command,  uint32_t tag
     return;
 
 fail:
-    pa_module_unload_request(u->module, true);
+    unload_module(u);
 }
 
 #else
@@ -1445,7 +1516,7 @@ static void source_info_cb(pa_pdispatch *pd, uint32_t command,  uint32_t tag, pa
     return;
 
 fail:
-    pa_module_unload_request(u->module, true);
+    unload_module(u);
 }
 
 #endif
@@ -1506,7 +1577,7 @@ static void command_subscribe_event(pa_pdispatch *pd,  uint32_t command,  uint32
     if (pa_tagstruct_getu32(t, &e) < 0 ||
         pa_tagstruct_getu32(t, &idx) < 0) {
         pa_log("Invalid protocol reply");
-        pa_module_unload_request(u->module, true);
+        unload_module(u);
         return;
     }
 
@@ -1653,7 +1724,7 @@ parse_error:
     pa_log("Invalid reply. (Create stream)");
 
 fail:
-    pa_module_unload_request(u->module, true);
+    unload_module(u);
 
 }
 
@@ -1857,7 +1928,7 @@ static void setup_complete_callback(pa_pdispatch *pd, uint32_t command, uint32_t
     return;
 
 fail:
-    pa_module_unload_request(u->module, true);
+    unload_module(u);
 }
 
 /* Called from main context */
@@ -1868,7 +1939,7 @@ static void pstream_die_callback(pa_pstream *p, void *userdata) {
     pa_assert(u);
 
     pa_log_warn("Stream died.");
-    pa_module_unload_request(u->module, true);
+    unload_module(u);
 }
 
 /* Called from main context */
@@ -1881,7 +1952,7 @@ static void pstream_packet_callback(pa_pstream *p, pa_packet *packet, pa_cmsg_an
 
     if (pa_pdispatch_run(u->pdispatch, packet, ancil_data, u) < 0) {
         pa_log("Invalid packet");
-        pa_module_unload_request(u->module, true);
+        unload_module(u);
         return;
     }
 }
@@ -1897,7 +1968,7 @@ static void pstream_memblock_callback(pa_pstream *p, uint32_t channel, int64_t o
 
     if (channel != u->channel) {
         pa_log("Received memory block on bad channel.");
-        pa_module_unload_request(u->module, true);
+        unload_module(u);
         return;
     }
 
@@ -1910,8 +1981,6 @@ static void pstream_memblock_callback(pa_pstream *p, uint32_t channel, int64_t o
 /* Called from main context */
 static void on_connection(pa_socket_client *sc, pa_iochannel *io, void *userdata) {
     struct userdata *u = userdata;
-    pa_tagstruct *t;
-    uint32_t tag;
 
     pa_assert(sc);
     pa_assert(u);
@@ -1922,11 +1991,31 @@ static void on_connection(pa_socket_client *sc, pa_iochannel *io, void *userdata
 
     if (!io) {
         pa_log("Connection failed: %s", pa_cstrerror(errno));
-        pa_module_unload_request(u->module, true);
+        unload_module(u);
         return;
     }
 
-    u->pstream = pa_pstream_new(u->core->mainloop, io, u->core->mempool);
+    u->io = io;
+
+#ifdef TUNNEL_SINK
+    pa_log_debug("Asking ctl thread to create sink.");
+    pa_asyncmsgq_post(u->thread_mq.outq, PA_MSGOBJECT(u->msg), TUNNEL_MESSAGE_CREATE_SINK_REQUEST, u, 0, NULL, NULL);
+#else
+    pa_log_debug("Asking ctl thread to create source.");
+    pa_asyncmsgq_post(u->thread_mq.outq, PA_MSGOBJECT(u->msg), TUNNEL_MESSAGE_CREATE_SOURCE_REQUEST, u, 0, NULL, NULL);
+#endif
+}
+
+#ifdef TUNNEL_SINK
+static void on_sink_created(struct userdata *u)
+#else
+static void on_source_created(struct userdata *u)
+#endif
+{
+    pa_tagstruct *t;
+    uint32_t tag;
+
+    u->pstream = pa_pstream_new(u->core->mainloop, u->io, u->core->mempool);
     u->pdispatch = pa_pdispatch_new(u->core->mainloop, true, command_table, PA_COMMAND_MAX);
 
     pa_pstream_set_die_callback(u->pstream, pstream_die_callback, u);
@@ -1946,8 +2035,8 @@ static void on_connection(pa_socket_client *sc, pa_iochannel *io, void *userdata
 {
     pa_creds ucred;
 
-    if (pa_iochannel_creds_supported(io))
-        pa_iochannel_creds_enable(io);
+    if (pa_iochannel_creds_supported(u->io))
+        pa_iochannel_creds_enable(u->io);
 
     ucred.uid = getuid();
     ucred.gid = getgid();
@@ -2004,25 +2093,207 @@ static void sink_set_mute(pa_sink *sink) {
 
 #endif
 
-int pa__init(pa_module*m) {
-    pa_modargs *ma = NULL;
-    struct userdata *u = NULL;
-    char *server = NULL;
-    pa_strlist *server_list = NULL;
-    pa_sample_spec ss;
-    pa_channel_map map;
-    char *dn = NULL;
-    uint32_t latency_msec;
 #ifdef TUNNEL_SINK
+static void create_sink(struct userdata *u) {
     pa_sink_new_data data;
+    char *data_name = NULL;
+
+    if (!(data_name = pa_xstrdup(u->configured_sink_name)))
+        data_name = pa_sprintf_malloc("tunnel-sink.%s", u->server_name);
+
+    pa_sink_new_data_init(&data);
+    data.driver = __FILE__;
+    data.module = u->module;
+    data.namereg_fail = false;
+    pa_sink_new_data_set_name(&data, data_name);
+    pa_sink_new_data_set_sample_spec(&data, &u->sample_spec);
+    pa_sink_new_data_set_channel_map(&data, &u->channel_map);
+    pa_proplist_setf(data.proplist, PA_PROP_DEVICE_DESCRIPTION, "%s%s%s", pa_strempty(u->sink_name), u->sink_name ? " on " : "", u->server_name);
+    pa_proplist_sets(data.proplist, "tunnel.remote.server", u->server_name);
+    if (u->sink_name)
+        pa_proplist_sets(data.proplist, "tunnel.remote.sink", u->sink_name);
+
+    pa_proplist_update(data.proplist, PA_UPDATE_REPLACE, u->sink_proplist);
+
+    u->sink = pa_sink_new(u->module->core, &data, PA_SINK_NETWORK|PA_SINK_LATENCY);
+
+    if (!u->sink) {
+        pa_log("Failed to create sink.");
+        goto finish;
+    }
+
+    u->sink->parent.process_msg = sink_process_msg;
+    u->sink->userdata = u;
+    u->sink->set_state_in_main_thread = sink_set_state_in_main_thread_cb;
+    pa_sink_set_set_volume_callback(u->sink, sink_set_volume);
+    pa_sink_set_set_mute_callback(u->sink, sink_set_mute);
+
+    u->sink->refresh_volume = u->sink->refresh_muted = false;
+
+/*     pa_sink_set_latency_range(u->sink, MIN_NETWORK_LATENCY_USEC, 0); */
+
+    pa_sink_set_asyncmsgq(u->sink, u->thread_mq.inq);
+    pa_sink_set_rtpoll(u->sink, u->rtpoll);
+    pa_sink_set_fixed_latency(u->sink, u->latency * PA_USEC_PER_MSEC);
+
+    pa_sink_put(u->sink);
+
+finish:
+    pa_sink_new_data_done(&data);
+    pa_xfree(data_name);
+
+    pa_asyncmsgq_post(u->sink->asyncmsgq, PA_MSGOBJECT(u->sink), SINK_MESSAGE_CREATED, u, 0, NULL, NULL);
+}
 #else
+static void create_source(struct userdata *u) {
     pa_source_new_data data;
+    char *data_name = NULL;
+
+    if (!(data_name = pa_xstrdup(u->configured_source_name)))
+        data_name = pa_sprintf_malloc("tunnel-source.%s", u->server_name);
+
+    pa_source_new_data_init(&data);
+    data.driver = __FILE__;
+    data.module = u->module;
+    data.namereg_fail = false;
+    pa_source_new_data_set_name(&data, data_name);
+    pa_source_new_data_set_sample_spec(&data, &u->sample_spec);
+    pa_source_new_data_set_channel_map(&data, &u->channel_map);
+    pa_proplist_setf(data.proplist, PA_PROP_DEVICE_DESCRIPTION, "%s%s%s", pa_strempty(u->source_name), u->source_name ? " on " : "", u->server_name);
+    pa_proplist_sets(data.proplist, "tunnel.remote.server", u->server_name);
+    if (u->source_name)
+        pa_proplist_sets(data.proplist, "tunnel.remote.source", u->source_name);
+
+    pa_proplist_update(data.proplist, PA_UPDATE_REPLACE, u->source_proplist);
+
+    u->source = pa_source_new(u->module->core, &data, PA_SOURCE_NETWORK|PA_SOURCE_LATENCY);
+
+    if (!u->source) {
+        pa_log("Failed to create source.");
+        goto finish;
+    }
+
+    u->source->parent.process_msg = source_process_msg;
+    u->source->set_state_in_main_thread = source_set_state_in_main_thread_cb;
+    u->source->userdata = u;
+
+/*     pa_source_set_latency_range(u->source, MIN_NETWORK_LATENCY_USEC, 0); */
+
+    pa_source_set_asyncmsgq(u->source, u->thread_mq.inq);
+    pa_source_set_rtpoll(u->source, u->rtpoll);
+    pa_source_set_fixed_latency(u->source, u->latency * PA_USEC_PER_MSEC);
+
+    u->mcalign = pa_mcalign_new(pa_frame_size(&u->source->sample_spec));
+
+    pa_source_put(u->source);
+
+finish:
+    pa_source_new_data_done(&data);
+    pa_xfree(data_name);
+
+    pa_asyncmsgq_post(u->source->asyncmsgq, PA_MSGOBJECT(u->source), SOURCE_MESSAGE_CREATED, u, 0, NULL, NULL);
+}
 #endif
+
+/* Runs in PA mainloop context */
+static int tunnel_process_msg(pa_msgobject *o, int code, void *data, int64_t offset, pa_memchunk *chunk) {
+    struct userdata *u = (struct userdata *) data;
+
+    pa_assert(u);
+    pa_assert_ctl_context();
+
+    if (u->shutting_down)
+        return 0;
+
+    switch (code) {
+#ifdef TUNNEL_SINK
+        case TUNNEL_MESSAGE_CREATE_SINK_REQUEST:
+            create_sink(u);
+            break;
+#else
+        case TUNNEL_MESSAGE_CREATE_SOURCE_REQUEST:
+            create_source(u);
+            break;
+#endif
+        case TUNNEL_MESSAGE_MAYBE_RESTART:
+            unload_module(u);
+            break;
+    }
+
+    return 0;
+}
+
+static int start_connect(struct userdata *u, char *server, bool automatic) {
+    pa_strlist *server_list = NULL;
+    int rc = 0;
+
+    if (server) {
+        if (!(server_list = pa_strlist_parse(server))) {
+            pa_log("Invalid server specified.");
+            rc = -1;
+            goto done;
+        }
+    } else {
+        char *ufn;
+
+        if (!automatic) {
+            pa_log("No server specified.");
+            rc = -1;
+            goto done;
+        }
+
+        pa_log("No server address found. Attempting default local sockets.");
+
+        /* The system wide instance via PF_LOCAL */
+        server_list = pa_strlist_prepend(server_list, PA_SYSTEM_RUNTIME_PATH PA_PATH_SEP PA_NATIVE_DEFAULT_UNIX_SOCKET);
+
+        /* The user instance via PF_LOCAL */
+        if ((ufn = pa_runtime_path(PA_NATIVE_DEFAULT_UNIX_SOCKET))) {
+            server_list = pa_strlist_prepend(server_list, ufn);
+            pa_xfree(ufn);
+        }
+    }
+
+    for (;;) {
+        server_list = pa_strlist_pop(server_list, &u->server_name);
+
+        if (!u->server_name) {
+            pa_log("Failed to connect to server '%s'", server);
+            rc = -1;
+            goto done;
+        }
+
+        pa_log_debug("Trying to connect to %s...", u->server_name);
+
+        if (!(u->client = pa_socket_client_new_string(u->module->core->mainloop, true, u->server_name, PA_NATIVE_DEFAULT_PORT))) {
+            pa_xfree(u->server_name);
+            u->server_name = NULL;
+            continue;
+        }
+
+        break;
+    }
+
+    if (u->client)
+        pa_socket_client_set_callback(u->client, on_connection, u);
+
+done:
+    pa_strlist_free(server_list);
+
+    return rc;
+}
+
+static int do_init(pa_module *m) {
+    pa_modargs *ma = NULL;
+    struct userdata *u = NULL;
+    char *server = NULL;
+    uint32_t latency_msec;
     bool automatic;
 #ifdef HAVE_X11
     xcb_connection_t *xcb = NULL;
 #endif
     const char *cookie_path;
+    uint32_t reconnect_interval_ms = 0;
 
     pa_assert(m);
 
@@ -2040,10 +2311,12 @@ int pa__init(pa_module*m) {
     u->server_name = NULL;
 #ifdef TUNNEL_SINK
     u->sink_name = pa_xstrdup(pa_modargs_get_value(ma, "sink", NULL));;
+    u->configured_sink_name = pa_xstrdup(pa_modargs_get_value(ma, "sink_name", NULL));
     u->sink = NULL;
     u->requested_bytes = 0;
 #else
     u->source_name = pa_xstrdup(pa_modargs_get_value(ma, "source", NULL));;
+    u->configured_source_name = pa_xstrdup(pa_modargs_get_value(ma, "source_name", NULL));
     u->source = NULL;
 #endif
 #ifndef USE_SMOOTHER_2
@@ -2066,6 +2339,9 @@ int pa__init(pa_module*m) {
     u->receive_snapshot = 0;
     u->receive_counter = 0;
 
+    u->msg = pa_msgobject_new(tunnel_msg);
+    u->msg->parent.process_msg = tunnel_process_msg;
+
     u->rtpoll = pa_rtpoll_new();
 
     if (pa_thread_mq_init(&u->thread_mq, m->core->mainloop, u->rtpoll) < 0) {
@@ -2172,151 +2448,37 @@ int pa__init(pa_module*m) {
             goto fail;
     }
 
-    if (server) {
-        if (!(server_list = pa_strlist_parse(server))) {
-            pa_log("Invalid server specified.");
-            goto fail;
-        }
-    } else {
-        char *ufn;
-
-        if (!automatic) {
-            pa_log("No server specified.");
-            goto fail;
-        }
-
-        pa_log("No server address found. Attempting default local sockets.");
-
-        /* The system wide instance via PF_LOCAL */
-        server_list = pa_strlist_prepend(server_list, PA_SYSTEM_RUNTIME_PATH PA_PATH_SEP PA_NATIVE_DEFAULT_UNIX_SOCKET);
-
-        /* The user instance via PF_LOCAL */
-        if ((ufn = pa_runtime_path(PA_NATIVE_DEFAULT_UNIX_SOCKET))) {
-            server_list = pa_strlist_prepend(server_list, ufn);
-            pa_xfree(ufn);
-        }
-    }
-
-    ss = m->core->default_sample_spec;
-    map = m->core->default_channel_map;
-    if (pa_modargs_get_sample_spec_and_channel_map(ma, &ss, &map, PA_CHANNEL_MAP_DEFAULT) < 0) {
+    u->sample_spec = m->core->default_sample_spec;
+    u->channel_map = m->core->default_channel_map;
+    if (pa_modargs_get_sample_spec_and_channel_map(ma, &u->sample_spec, &u->channel_map, PA_CHANNEL_MAP_DEFAULT) < 0) {
         pa_log("Invalid sample format specification");
         goto fail;
     }
 
 #ifdef USE_SMOOTHER_2
     /* Smoother window must be larger than time between updates. */
-    u->smoother = pa_smoother_2_new(LATENCY_INTERVAL + 5*PA_USEC_PER_SEC, pa_rtclock_now(), pa_frame_size(&ss), ss.rate);
+    u->smoother = pa_smoother_2_new(LATENCY_INTERVAL + 5*PA_USEC_PER_SEC, pa_rtclock_now(), pa_frame_size(&u->sample_spec), u->sample_spec.rate);
 #endif
 
-    for (;;) {
-        server_list = pa_strlist_pop(server_list, &u->server_name);
-
-        if (!u->server_name) {
-            pa_log("Failed to connect to server '%s'", server);
-            goto fail;
-        }
-
-        pa_log_debug("Trying to connect to %s...", u->server_name);
-
-        if (!(u->client = pa_socket_client_new_string(m->core->mainloop, true, u->server_name, PA_NATIVE_DEFAULT_PORT))) {
-            pa_xfree(u->server_name);
-            u->server_name = NULL;
-            continue;
-        }
-
-        break;
-     }
-
-    pa_socket_client_set_callback(u->client, on_connection, u);
+    pa_modargs_get_value_u32(ma, "reconnect_interval_ms", &reconnect_interval_ms);
+    u->reconnect_interval_us = reconnect_interval_ms * PA_USEC_PER_MSEC;
 
 #ifdef TUNNEL_SINK
 
-    if (!(dn = pa_xstrdup(pa_modargs_get_value(ma, "sink_name", NULL))))
-        dn = pa_sprintf_malloc("tunnel-sink.%s", u->server_name);
-
-    pa_sink_new_data_init(&data);
-    data.driver = __FILE__;
-    data.module = m;
-    data.namereg_fail = false;
-    pa_sink_new_data_set_name(&data, dn);
-    pa_sink_new_data_set_sample_spec(&data, &ss);
-    pa_sink_new_data_set_channel_map(&data, &map);
-    pa_proplist_setf(data.proplist, PA_PROP_DEVICE_DESCRIPTION, "%s%s%s", pa_strempty(u->sink_name), u->sink_name ? " on " : "", u->server_name);
-    pa_proplist_sets(data.proplist, "tunnel.remote.server", u->server_name);
-    if (u->sink_name)
-        pa_proplist_sets(data.proplist, "tunnel.remote.sink", u->sink_name);
-
-    if (pa_modargs_get_proplist(ma, "sink_properties", data.proplist, PA_UPDATE_REPLACE) < 0) {
+    u->sink_proplist = pa_proplist_new();
+    if (pa_modargs_get_proplist(ma, "sink_properties", u->sink_proplist, PA_UPDATE_REPLACE) < 0) {
         pa_log("Invalid properties");
-        pa_sink_new_data_done(&data);
         goto fail;
     }
 
-    u->sink = pa_sink_new(m->core, &data, PA_SINK_NETWORK|PA_SINK_LATENCY);
-    pa_sink_new_data_done(&data);
-
-    if (!u->sink) {
-        pa_log("Failed to create sink.");
-        goto fail;
-    }
-
-    u->sink->parent.process_msg = sink_process_msg;
-    u->sink->userdata = u;
-    u->sink->set_state_in_main_thread = sink_set_state_in_main_thread_cb;
-    pa_sink_set_set_volume_callback(u->sink, sink_set_volume);
-    pa_sink_set_set_mute_callback(u->sink, sink_set_mute);
-
-    u->sink->refresh_volume = u->sink->refresh_muted = false;
-
-/*     pa_sink_set_latency_range(u->sink, MIN_NETWORK_LATENCY_USEC, 0); */
-
-    pa_sink_set_asyncmsgq(u->sink, u->thread_mq.inq);
-    pa_sink_set_rtpoll(u->sink, u->rtpoll);
-    pa_sink_set_fixed_latency(u->sink, latency_msec * PA_USEC_PER_MSEC);
-
 #else
 
-    if (!(dn = pa_xstrdup(pa_modargs_get_value(ma, "source_name", NULL))))
-        dn = pa_sprintf_malloc("tunnel-source.%s", u->server_name);
-
-    pa_source_new_data_init(&data);
-    data.driver = __FILE__;
-    data.module = m;
-    data.namereg_fail = false;
-    pa_source_new_data_set_name(&data, dn);
-    pa_source_new_data_set_sample_spec(&data, &ss);
-    pa_source_new_data_set_channel_map(&data, &map);
-    pa_proplist_setf(data.proplist, PA_PROP_DEVICE_DESCRIPTION, "%s%s%s", pa_strempty(u->source_name), u->source_name ? " on " : "", u->server_name);
-    pa_proplist_sets(data.proplist, "tunnel.remote.server", u->server_name);
-    if (u->source_name)
-        pa_proplist_sets(data.proplist, "tunnel.remote.source", u->source_name);
-
-    if (pa_modargs_get_proplist(ma, "source_properties", data.proplist, PA_UPDATE_REPLACE) < 0) {
+    u->source_proplist = pa_proplist_new();
+    if (pa_modargs_get_proplist(ma, "source_properties", u->source_proplist, PA_UPDATE_REPLACE) < 0) {
         pa_log("Invalid properties");
-        pa_source_new_data_done(&data);
         goto fail;
     }
 
-    u->source = pa_source_new(m->core, &data, PA_SOURCE_NETWORK|PA_SOURCE_LATENCY);
-    pa_source_new_data_done(&data);
-
-    if (!u->source) {
-        pa_log("Failed to create source.");
-        goto fail;
-    }
-
-    u->source->parent.process_msg = source_process_msg;
-    u->source->set_state_in_main_thread = source_set_state_in_main_thread_cb;
-    u->source->userdata = u;
-
-/*     pa_source_set_latency_range(u->source, MIN_NETWORK_LATENCY_USEC, 0); */
-
-    pa_source_set_asyncmsgq(u->source, u->thread_mq.inq);
-    pa_source_set_rtpoll(u->source, u->rtpoll);
-    pa_source_set_fixed_latency(u->source, latency_msec * PA_USEC_PER_MSEC);
-
-    u->mcalign = pa_mcalign_new(pa_frame_size(&u->source->sample_spec));
 #endif
 
     u->time_event = NULL;
@@ -2328,25 +2490,18 @@ int pa__init(pa_module*m) {
     u->fragsize = (uint32_t) -1;
 #endif
 
+    if (start_connect(u, server, automatic) < 0) {
+        goto fail;
+    }
+
     if (!(u->thread = pa_thread_new("module-tunnel", thread_func, u))) {
         pa_log("Failed to create thread.");
         goto fail;
     }
 
-#ifdef TUNNEL_SINK
-    pa_sink_put(u->sink);
-#else
-    pa_source_put(u->source);
-#endif
-
-    pa_xfree(dn);
-
     if (server)
         pa_xfree(server);
 
-    if (server_list)
-        pa_strlist_free(server_list);
-
 #ifdef HAVE_X11
     if (xcb)
         xcb_disconnect(xcb);
@@ -2357,14 +2512,9 @@ int pa__init(pa_module*m) {
     return 0;
 
 fail:
-    pa__done(m);
-
     if (server)
         pa_xfree(server);
 
-    if (server_list)
-        pa_strlist_free(server_list);
-
 #ifdef HAVE_X11
     if (xcb)
         xcb_disconnect(xcb);
@@ -2373,19 +2523,19 @@ fail:
     if (ma)
         pa_modargs_free(ma);
 
-    pa_xfree(dn);
-
     return -1;
 }
 
-void pa__done(pa_module*m) {
-    struct userdata* u;
+static void do_done(pa_module *m) {
+    struct userdata *u = NULL;
 
     pa_assert(m);
 
     if (!(u = m->userdata))
         return;
 
+    u->shutting_down = true;
+
 #ifdef TUNNEL_SINK
     if (u->sink)
         pa_sink_unlink(u->sink);
@@ -2443,8 +2593,12 @@ void pa__done(pa_module*m) {
 
 #ifdef TUNNEL_SINK
     pa_xfree(u->sink_name);
+    pa_xfree(u->configured_sink_name);
+    pa_proplist_free(u->sink_proplist);
 #else
     pa_xfree(u->source_name);
+    pa_xfree(u->configured_source_name);
+    pa_proplist_free(u->source_proplist);
 #endif
     pa_xfree(u->server_name);
 
@@ -2452,5 +2606,28 @@ void pa__done(pa_module*m) {
     pa_xfree(u->server_fqdn);
     pa_xfree(u->user_name);
 
+    pa_xfree(u->msg);
+
     pa_xfree(u);
+
+    m->userdata = NULL;
+}
+
+int pa__init(pa_module *m) {
+    int ret;
+
+    pa_assert(m);
+
+    ret = do_init(m);
+
+    if (ret < 0)
+        pa__done(m);
+
+    return ret;
+}
+
+void pa__done(pa_module *m) {
+    pa_assert(m);
+
+    do_done(m);
 }