diff options
Diffstat (limited to 'src/modules/module-tunnel.c')
-rw-r--r-- | src/modules/module-tunnel.c | 539 |
1 files changed, 358 insertions, 181 deletions
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); } |