mirrors/pulseaudio
1
0
Fork 0
mirror of https://gitlab.freedesktop.org/pulseaudio/pulseaudio.git synced 2025-11-05 13:29:57 -05:00
Code Issues Projects Releases Packages Wiki Activity Actions

module-tunnel: restart module

Browse source 
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>
This commit is contained in:
Craig Howard 2022-05-06 16:05:33 -07:00 • committed by PulseAudio Marge Bot
parent f81bb09772
commit 17eb178f17
2 changed files with 361 additions and 184 deletions
Download patch file Download diff file Expand all files Collapse all files

4
src/modules/meson.build
View file

@ -54,9 +54,9 @@ all_modules = [
[ 'module-suspend-on-idle', 'module-suspend-on-idle.c' ], [ 'module-suspend-on-idle', 'module-suspend-on-idle.c' ],
[ 'module-switch-on-connect', 'module-switch-on-connect.c' ], [ 'module-switch-on-connect', 'module-switch-on-connect.c' ],
[ 'module-switch-on-port-available', 'module-switch-on-port-available.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-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-tunnel-source-new', ['module-tunnel-source-new.c', 'restart-module.c'] ],
[ 'module-virtual-sink', 'module-virtual-sink.c' ], [ 'module-virtual-sink', 'module-virtual-sink.c' ],
[ 'module-virtual-source', 'module-virtual-source.c' ], [ 'module-virtual-source', 'module-virtual-source.c' ],

541
src/modules/module-tunnel.c
View file

@ -22,6 +22,8 @@
#include <config.h> #include <config.h>
#endif #endif
#include "restart-module.h"
#include <unistd.h> #include <unistd.h>
#include <string.h> #include <string.h>
#include <errno.h> #include <errno.h>
@ -81,6 +83,7 @@ PA_MODULE_USAGE(
"auto=<determine server/sink/cookie automatically> " "auto=<determine server/sink/cookie automatically> "
"server=<address> " "server=<address> "
"sink=<remote sink name> " "sink=<remote sink name> "
"reconnect_interval_ms=<interval to try reconnects, 0 or omitted if disabled> "
"cookie=<filename> " "cookie=<filename> "
"format=<sample format> " "format=<sample format> "
"channels=<number of channels> " "channels=<number of channels> "
@ -95,6 +98,7 @@ PA_MODULE_USAGE(
"auto=<determine server/source/cookie automatically> " "auto=<determine server/source/cookie automatically> "
"server=<address> " "server=<address> "
"source=<remote source name> " "source=<remote source name> "
"reconnect_interval_ms=<interval to try reconnects, 0 or omitted if disabled> "
"cookie=<filename> " "cookie=<filename> "
"format=<sample format> " "format=<sample format> "
"channels=<number of channels> " "channels=<number of channels> "
@ -115,6 +119,7 @@ static const char* const valid_modargs[] = {
"channels", "channels",
"rate", "rate",
"latency_msec", "latency_msec",
"reconnect_interval_ms",
#ifdef TUNNEL_SINK #ifdef TUNNEL_SINK
"sink_name", "sink_name",
"sink_properties", "sink_properties",
@ -141,7 +146,8 @@ enum {
SINK_MESSAGE_REMOTE_SUSPEND, SINK_MESSAGE_REMOTE_SUSPEND,
SINK_MESSAGE_UPDATE_LATENCY, SINK_MESSAGE_UPDATE_LATENCY,
SINK_MESSAGE_GET_LATENCY_SNAPSHOT, SINK_MESSAGE_GET_LATENCY_SNAPSHOT,
SINK_MESSAGE_POST SINK_MESSAGE_POST,
SINK_MESSAGE_CREATED
}; };
#define DEFAULT_LATENCY_MSEC 100 #define DEFAULT_LATENCY_MSEC 100
@ -152,13 +158,33 @@ enum {
SOURCE_MESSAGE_POST = PA_SOURCE_MESSAGE_MAX, SOURCE_MESSAGE_POST = PA_SOURCE_MESSAGE_MAX,
SOURCE_MESSAGE_REMOTE_SUSPEND, SOURCE_MESSAGE_REMOTE_SUSPEND,
SOURCE_MESSAGE_UPDATE_LATENCY, SOURCE_MESSAGE_UPDATE_LATENCY,
SOURCE_MESSAGE_GET_LATENCY_SNAPSHOT SOURCE_MESSAGE_GET_LATENCY_SNAPSHOT,
SOURCE_MESSAGE_CREATED
}; };
#define DEFAULT_LATENCY_MSEC 25 #define DEFAULT_LATENCY_MSEC 25
#endif #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 #ifdef TUNNEL_SINK
static void command_request(pa_pdispatch *pd, uint32_t command, uint32_t tag, pa_tagstruct *t, void *userdata); 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); 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; char *server_name;
#ifdef TUNNEL_SINK #ifdef TUNNEL_SINK
char *sink_name; char *sink_name;
char *configured_sink_name;
pa_sink *sink; pa_sink *sink;
size_t requested_bytes; size_t requested_bytes;
#else #else
char *source_name; char *source_name;
char *configured_source_name;
pa_source *source; pa_source *source;
pa_mcalign *mcalign; pa_mcalign *mcalign;
#endif #endif
@ -229,6 +257,7 @@ struct userdata {
bool remote_corked:1; bool remote_corked:1;
bool remote_suspended:1; bool remote_suspended:1;
bool shutting_down:1;
pa_usec_t transport_usec; /* maintained in the main thread */ pa_usec_t transport_usec; /* maintained in the main thread */
pa_usec_t thread_transport_usec; /* maintained in the IO thread */ pa_usec_t thread_transport_usec; /* maintained in the IO thread */
@ -252,12 +281,42 @@ struct userdata {
uint32_t tlength; uint32_t tlength;
uint32_t minreq; uint32_t minreq;
uint32_t prebuf; uint32_t prebuf;
pa_proplist *sink_proplist;
#else #else
uint32_t fragsize; uint32_t fragsize;
pa_proplist *source_proplist;
#endif #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); 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 */ /* 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) { 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_assert(u->pdispatch == pd);
pa_log_warn("Stream killed"); pa_log_warn("Stream killed");
pa_module_unload_request(u->module, true); unload_module(u);
} }
/* Called from main context */ /* 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_tagstruct_eof(t)) {
pa_log("Invalid packet."); pa_log("Invalid packet.");
pa_module_unload_request(u->module, true); unload_module(u);
return; 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_tagstruct_get_boolean(t, &suspended) < 0) {
pa_log_error("Invalid packet."); pa_log_error("Invalid packet.");
pa_module_unload_request(u->module, true); unload_module(u);
return; 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_tagstruct_getu32(t, &maxlength) < 0) {
pa_log_error("Invalid packet."); pa_log_error("Invalid packet.");
pa_module_unload_request(u->module, true); unload_module(u);
return; 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_tagstruct_get_usec(t, &usec) < 0) {
pa_log_error("Invalid packet."); pa_log_error("Invalid packet.");
pa_module_unload_request(u->module, true); unload_module(u);
return; return;
} }
} else { } 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_tagstruct_get_usec(t, &usec) < 0) {
pa_log_error("Invalid packet."); pa_log_error("Invalid packet.");
pa_module_unload_request(u->module, true); unload_module(u);
return; return;
} }
} }
@ -612,6 +671,12 @@ static int sink_process_msg(pa_msgobject *o, int code, void *data, int64_t offse
u->receive_counter += chunk->length; u->receive_counter += chunk->length;
return 0;
case SINK_MESSAGE_CREATED:
on_sink_created(u);
return 0; return 0;
} }
@ -740,6 +805,12 @@ static int source_process_msg(pa_msgobject *o, int code, void *data, int64_t off
return 0; return 0;
} }
case SOURCE_MESSAGE_CREATED:
on_source_created(u);
return 0;
} }
return pa_source_process_msg(o, code, data, offset, chunk); return pa_source_process_msg(o, code, data, offset, chunk);
@ -793,7 +864,7 @@ static void thread_func(void *userdata) {
int ret; int ret;
#ifdef TUNNEL_SINK #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); pa_sink_process_rewind(u->sink, 0);
#endif #endif
@ -807,7 +878,7 @@ static void thread_func(void *userdata) {
fail: fail:
/* If this was no regular exit from the loop we have to continue /* If this was no regular exit from the loop we have to continue
* processing messages until we received PA_MESSAGE_SHUTDOWN */ * 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); pa_asyncmsgq_wait_for(u->thread_mq.inq, PA_MESSAGE_SHUTDOWN);
finish: finish:
@ -841,7 +912,7 @@ static void command_request(pa_pdispatch *pd, uint32_t command, uint32_t tag, p
return; return;
fail: fail:
pa_module_unload_request(u->module, true); unload_module(u);
} }
#endif #endif
@ -956,7 +1027,7 @@ static void stream_get_latency_callback(pa_pdispatch *pd, uint32_t command, uint
fail: fail:
pa_module_unload_request(u->module, true); unload_module(u);
} }
/* Called from main context */ /* Called from main context */
@ -1091,7 +1162,7 @@ static void server_info_cb(pa_pdispatch *pd, uint32_t command, uint32_t tag, pa
return; return;
fail: fail:
pa_module_unload_request(u->module, true); unload_module(u);
} }
static int read_ports(struct userdata *u, pa_tagstruct *t) { 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; return;
fail: fail:
pa_module_unload_request(u->module, true); unload_module(u);
} }
/* Called from main context */ /* Called from main context */
@ -1355,7 +1426,7 @@ static void sink_input_info_cb(pa_pdispatch *pd, uint32_t command, uint32_t tag
return; return;
fail: fail:
pa_module_unload_request(u->module, true); unload_module(u);
} }
#else #else
@ -1445,7 +1516,7 @@ static void source_info_cb(pa_pdispatch *pd, uint32_t command, uint32_t tag, pa
return; return;
fail: fail:
pa_module_unload_request(u->module, true); unload_module(u);
} }
#endif #endif
@ -1506,7 +1577,7 @@ static void command_subscribe_event(pa_pdispatch *pd, uint32_t command, uint32
if (pa_tagstruct_getu32(t, &e) < 0 || if (pa_tagstruct_getu32(t, &e) < 0 ||
pa_tagstruct_getu32(t, &idx) < 0) { pa_tagstruct_getu32(t, &idx) < 0) {
pa_log("Invalid protocol reply"); pa_log("Invalid protocol reply");
pa_module_unload_request(u->module, true); unload_module(u);
return; return;
} }
@ -1653,7 +1724,7 @@ parse_error:
pa_log("Invalid reply. (Create stream)"); pa_log("Invalid reply. (Create stream)");
fail: 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; return;
fail: fail:
pa_module_unload_request(u->module, true); unload_module(u);
} }
/* Called from main context */ /* Called from main context */
@ -1868,7 +1939,7 @@ static void pstream_die_callback(pa_pstream *p, void *userdata) {
pa_assert(u); pa_assert(u);
pa_log_warn("Stream died."); pa_log_warn("Stream died.");
pa_module_unload_request(u->module, true); unload_module(u);
} }
/* Called from main context */ /* 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) { if (pa_pdispatch_run(u->pdispatch, packet, ancil_data, u) < 0) {
pa_log("Invalid packet"); pa_log("Invalid packet");
pa_module_unload_request(u->module, true); unload_module(u);
return; return;
} }
} }
@ -1897,7 +1968,7 @@ static void pstream_memblock_callback(pa_pstream *p, uint32_t channel, int64_t o
if (channel != u->channel) { if (channel != u->channel) {
pa_log("Received memory block on bad channel."); pa_log("Received memory block on bad channel.");
pa_module_unload_request(u->module, true); unload_module(u);
return; return;
} }
@ -1910,8 +1981,6 @@ static void pstream_memblock_callback(pa_pstream *p, uint32_t channel, int64_t o
/* Called from main context */ /* Called from main context */
static void on_connection(pa_socket_client *sc, pa_iochannel *io, void *userdata) { static void on_connection(pa_socket_client *sc, pa_iochannel *io, void *userdata) {
struct userdata *u = userdata; struct userdata *u = userdata;
pa_tagstruct *t;
uint32_t tag;
pa_assert(sc); pa_assert(sc);
pa_assert(u); pa_assert(u);
@ -1922,11 +1991,31 @@ static void on_connection(pa_socket_client *sc, pa_iochannel *io, void *userdata
if (!io) { if (!io) {
pa_log("Connection failed: %s", pa_cstrerror(errno)); pa_log("Connection failed: %s", pa_cstrerror(errno));
pa_module_unload_request(u->module, true); unload_module(u);
return; 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); 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); 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; pa_creds ucred;
if (pa_iochannel_creds_supported(io)) if (pa_iochannel_creds_supported(u->io))
pa_iochannel_creds_enable(io); pa_iochannel_creds_enable(u->io);
ucred.uid = getuid(); ucred.uid = getuid();
ucred.gid = getgid(); ucred.gid = getgid();
@ -2004,25 +2093,207 @@ static void sink_set_mute(pa_sink *sink) {
#endif #endif
int pa__init(pa_module*m) { #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; pa_modargs *ma = NULL;
struct userdata *u = NULL; struct userdata *u = NULL;
char *server = NULL; char *server = NULL;
pa_strlist *server_list = NULL;
pa_sample_spec ss;
pa_channel_map map;
char *dn = NULL;
uint32_t latency_msec; uint32_t latency_msec;
#ifdef TUNNEL_SINK
pa_sink_new_data data;
#else
pa_source_new_data data;
#endif
bool automatic; bool automatic;
#ifdef HAVE_X11 #ifdef HAVE_X11
xcb_connection_t *xcb = NULL; xcb_connection_t *xcb = NULL;
#endif #endif
const char *cookie_path; const char *cookie_path;
uint32_t reconnect_interval_ms = 0;
pa_assert(m); pa_assert(m);
@ -2040,10 +2311,12 @@ int pa__init(pa_module*m) {
u->server_name = NULL; u->server_name = NULL;
#ifdef TUNNEL_SINK #ifdef TUNNEL_SINK
u->sink_name = pa_xstrdup(pa_modargs_get_value(ma, "sink", NULL));; 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->sink = NULL;
u->requested_bytes = 0; u->requested_bytes = 0;
#else #else
u->source_name = pa_xstrdup(pa_modargs_get_value(ma, "source", NULL));; 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; u->source = NULL;
#endif #endif
#ifndef USE_SMOOTHER_2 #ifndef USE_SMOOTHER_2
@ -2066,6 +2339,9 @@ int pa__init(pa_module*m) {
u->receive_snapshot = 0; u->receive_snapshot = 0;
u->receive_counter = 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(); u->rtpoll = pa_rtpoll_new();
if (pa_thread_mq_init(&u->thread_mq, m->core->mainloop, u->rtpoll) < 0) { 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; goto fail;
} }
if (server) { u->sample_spec = m->core->default_sample_spec;
if (!(server_list = pa_strlist_parse(server))) { u->channel_map = m->core->default_channel_map;
pa_log("Invalid server specified."); if (pa_modargs_get_sample_spec_and_channel_map(ma, &u->sample_spec, &u->channel_map, PA_CHANNEL_MAP_DEFAULT) < 0) {
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) {
pa_log("Invalid sample format specification"); pa_log("Invalid sample format specification");
goto fail; goto fail;
} }
#ifdef USE_SMOOTHER_2 #ifdef USE_SMOOTHER_2
/* Smoother window must be larger than time between updates. */ /* 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 #endif
for (;;) { pa_modargs_get_value_u32(ma, "reconnect_interval_ms", &reconnect_interval_ms);
server_list = pa_strlist_pop(server_list, &u->server_name); u->reconnect_interval_us = reconnect_interval_ms * PA_USEC_PER_MSEC;
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);
#ifdef TUNNEL_SINK #ifdef TUNNEL_SINK
if (!(dn = pa_xstrdup(pa_modargs_get_value(ma, "sink_name", NULL)))) u->sink_proplist = pa_proplist_new();
dn = pa_sprintf_malloc("tunnel-sink.%s", u->server_name); if (pa_modargs_get_proplist(ma, "sink_properties", u->sink_proplist, PA_UPDATE_REPLACE) < 0) {
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) {
pa_log("Invalid properties"); pa_log("Invalid properties");
pa_sink_new_data_done(&data);
goto fail; 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 #else
if (!(dn = pa_xstrdup(pa_modargs_get_value(ma, "source_name", NULL)))) u->source_proplist = pa_proplist_new();
dn = pa_sprintf_malloc("tunnel-source.%s", u->server_name); if (pa_modargs_get_proplist(ma, "source_properties", u->source_proplist, PA_UPDATE_REPLACE) < 0) {
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) {
pa_log("Invalid properties"); pa_log("Invalid properties");
pa_source_new_data_done(&data);
goto fail; 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 #endif
u->time_event = NULL; u->time_event = NULL;
@ -2328,25 +2490,18 @@ int pa__init(pa_module*m) {
u->fragsize = (uint32_t) -1; u->fragsize = (uint32_t) -1;
#endif #endif
if (start_connect(u, server, automatic) < 0) {
goto fail;
}
if (!(u->thread = pa_thread_new("module-tunnel", thread_func, u))) { if (!(u->thread = pa_thread_new("module-tunnel", thread_func, u))) {
pa_log("Failed to create thread."); pa_log("Failed to create thread.");
goto fail; goto fail;
} }
#ifdef TUNNEL_SINK
pa_sink_put(u->sink);
#else
pa_source_put(u->source);
#endif
pa_xfree(dn);
if (server) if (server)
pa_xfree(server); pa_xfree(server);
if (server_list)
pa_strlist_free(server_list);
#ifdef HAVE_X11 #ifdef HAVE_X11
if (xcb) if (xcb)
xcb_disconnect(xcb); xcb_disconnect(xcb);
@ -2357,14 +2512,9 @@ int pa__init(pa_module*m) {
return 0; return 0;
fail: fail:
pa__done(m);
if (server) if (server)
pa_xfree(server); pa_xfree(server);
if (server_list)
pa_strlist_free(server_list);
#ifdef HAVE_X11 #ifdef HAVE_X11
if (xcb) if (xcb)
xcb_disconnect(xcb); xcb_disconnect(xcb);
@ -2373,19 +2523,19 @@ fail:
if (ma) if (ma)
pa_modargs_free(ma); pa_modargs_free(ma);
pa_xfree(dn);
return -1; return -1;
} }
void pa__done(pa_module*m) { static void do_done(pa_module *m) {
struct userdata* u; struct userdata *u = NULL;
pa_assert(m); pa_assert(m);
if (!(u = m->userdata)) if (!(u = m->userdata))
return; return;
u->shutting_down = true;
#ifdef TUNNEL_SINK #ifdef TUNNEL_SINK
if (u->sink) if (u->sink)
pa_sink_unlink(u->sink); pa_sink_unlink(u->sink);
@ -2443,8 +2593,12 @@ void pa__done(pa_module*m) {
#ifdef TUNNEL_SINK #ifdef TUNNEL_SINK
pa_xfree(u->sink_name); pa_xfree(u->sink_name);
pa_xfree(u->configured_sink_name);
pa_proplist_free(u->sink_proplist);
#else #else
pa_xfree(u->source_name); pa_xfree(u->source_name);
pa_xfree(u->configured_source_name);
pa_proplist_free(u->source_proplist);
#endif #endif
pa_xfree(u->server_name); pa_xfree(u->server_name);
@ -2452,5 +2606,28 @@ void pa__done(pa_module*m) {
pa_xfree(u->server_fqdn); pa_xfree(u->server_fqdn);
pa_xfree(u->user_name); pa_xfree(u->user_name);
pa_xfree(u->msg);
pa_xfree(u); 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);
} }