[pulseaudio-commits] [Git][pulseaudio/pulseaudio][master] 5 commits: tunnel-sink-new: refactor sink creation
PulseAudio Marge Bot (@pulseaudio-merge-bot)
gitlab at gitlab.freedesktop.org
Mon May 16 18:28:31 UTC 2022
PulseAudio Marge Bot pushed to branch master at PulseAudio / pulseaudio
Commits:
117fa0cb by Craig Howard at 2022-05-16T18:26:16+00:00
tunnel-sink-new: refactor sink creation
Move the sink creation logic to its own function. This is in
preparation for sinks being created async. Store the relevant config
parameters in userdata, so create_sink() can access that data.
Part-of: <https://gitlab.freedesktop.org/pulseaudio/pulseaudio/-/merge_requests/688>
- - - - -
34d00afc by Craig Howard at 2022-05-16T18:26:16+00:00
tunnel-sink-new: create sink *after* connection
The io thread, after connection, sends a message asking for a sink to be
created. After the ctl thread is done with creation, it sends a message
back to the io thread so it can continue. This ensures that the sink
only exists when it's connected to something.
Part-of: <https://gitlab.freedesktop.org/pulseaudio/pulseaudio/-/merge_requests/688>
- - - - -
a48bee4f by Craig Howard at 2022-05-16T18:26:16+00:00
tunnel-sink-new: reinit module
When configured, reinitialize the module instead of exiting. This
allows a restart/reconnect, but the module to appear to always be alive
when the user does: "pactl list modules". (The sink will still not
exist until the tcp connection is established.)
Part-of: <https://gitlab.freedesktop.org/pulseaudio/pulseaudio/-/merge_requests/688>
- - - - -
f81bb097 by Craig Howard at 2022-05-16T18:26:16+00:00
tunnel-source-new: add restart/reconnect logic
Applies the same changes as to tunnel-sink-new: defer the source
creation until after the TCP connection is restablished, then attempt to
restart the module on failure.
Part-of: <https://gitlab.freedesktop.org/pulseaudio/pulseaudio/-/merge_requests/688>
- - - - -
17eb178f by Craig Howard at 2022-05-16T18:26:16+00:00
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>
- - - - -
6 changed files:
- src/modules/meson.build
- src/modules/module-tunnel-sink-new.c
- src/modules/module-tunnel-source-new.c
- src/modules/module-tunnel.c
- + src/modules/restart-module.c
- + src/modules/restart-module.h
Changes:
=====================================
src/modules/meson.build
=====================================
@@ -54,10 +54,10 @@ 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-new', 'module-tunnel-sink-new.c' ],
- [ 'module-tunnel-source', 'module-tunnel.c', [], [], [x11_dep] ],
- [ 'module-tunnel-source-new', 'module-tunnel-source-new.c' ],
+ [ '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', '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' ],
[ 'module-volume-restore', 'module-volume-restore.c' ],
=====================================
src/modules/module-tunnel-sink-new.c
=====================================
@@ -21,6 +21,8 @@
#include <config.h>
#endif
+#include "restart-module.h"
+
#include <pulse/context.h>
#include <pulse/timeval.h>
#include <pulse/xmalloc.h>
@@ -50,6 +52,7 @@ PA_MODULE_USAGE(
"sink=<name of the remote sink> "
"sink_name=<name for the local sink> "
"sink_properties=<properties for the local sink> "
+ "reconnect_interval_ms=<interval to try reconnects, 0 or omitted if disabled> "
"format=<sample format> "
"channels=<number of channels> "
"rate=<sample rate> "
@@ -60,12 +63,30 @@ PA_MODULE_USAGE(
#define MAX_LATENCY_USEC (200 * PA_USEC_PER_MSEC)
#define TUNNEL_THREAD_FAILED_MAINLOOP 1
+static int do_init(pa_module *m);
+static void do_done(pa_module *m);
static void stream_state_cb(pa_stream *stream, void *userdata);
static void stream_changed_buffer_attr_cb(pa_stream *stream, void *userdata);
static void stream_set_buffer_attr_cb(pa_stream *stream, int success, void *userdata);
static void context_state_cb(pa_context *c, void *userdata);
static void sink_update_requested_latency_cb(pa_sink *s);
+struct tunnel_msg {
+ pa_msgobject parent;
+};
+
+typedef struct tunnel_msg tunnel_msg;
+PA_DEFINE_PRIVATE_CLASS(tunnel_msg, pa_msgobject);
+
+enum {
+ TUNNEL_MESSAGE_CREATE_SINK_REQUEST,
+ TUNNEL_MESSAGE_MAYBE_RESTART,
+};
+
+enum {
+ TUNNEL_MESSAGE_SINK_CREATED = PA_SINK_MESSAGE_MAX,
+};
+
struct userdata {
pa_module *module;
pa_sink *sink;
@@ -81,10 +102,20 @@ struct userdata {
bool update_stream_bufferattr_after_connect;
bool connected;
+ bool shutting_down;
char *cookie_file;
char *remote_server;
char *remote_sink_name;
+ char *sink_name;
+
+ pa_proplist *sink_proplist;
+ pa_sample_spec sample_spec;
+ pa_channel_map channel_map;
+
+ tunnel_msg *msg;
+
+ pa_usec_t reconnect_interval_us;
};
static const char* const valid_modargs[] = {
@@ -97,7 +128,7 @@ static const char* const valid_modargs[] = {
"rate",
"channel_map",
"cookie",
- /* "reconnect", reconnect if server comes back again - unimplemented */
+ "reconnect_interval_ms",
NULL,
};
@@ -143,6 +174,7 @@ static pa_proplist* tunnel_new_proplist(struct userdata *u) {
static void thread_func(void *userdata) {
struct userdata *u = userdata;
pa_proplist *proplist;
+
pa_assert(u);
pa_log_debug("Thread starting up");
@@ -169,7 +201,7 @@ static void thread_func(void *userdata) {
u->remote_server,
PA_CONTEXT_NOAUTOSPAWN,
NULL) < 0) {
- pa_log("Failed to connect libpulse context");
+ pa_log("Failed to connect libpulse context: %s", pa_strerror(pa_context_errno(u->context)));
goto fail;
}
@@ -183,7 +215,7 @@ static void thread_func(void *userdata) {
goto fail;
}
- 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);
if (u->connected &&
@@ -221,7 +253,10 @@ static void thread_func(void *userdata) {
}
}
fail:
- pa_asyncmsgq_post(u->thread_mq->outq, PA_MSGOBJECT(u->module->core), PA_CORE_MESSAGE_UNLOAD_MODULE, u->module, 0, NULL, NULL);
+ /* send a message to the ctl thread to ask it to either terminate us, or
+ * restart us, but either way this thread will exit, so then wait for the
+ * shutdown message */
+ 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:
@@ -300,6 +335,78 @@ static void stream_overflow_callback(pa_stream *stream, void *userdata) {
pa_log_info("Server signalled buffer overrun.");
}
+/* Do a reinit of the module. Note that u will be freed as a result of this
+ * call. */
+static void maybe_restart(struct userdata *u) {
+ if (u->reconnect_interval_us > 0) {
+ pa_restart_module_reinit(u->module, do_init, do_done, u->reconnect_interval_us);
+ } else {
+ /* exit the module */
+ pa_asyncmsgq_post(u->thread_mq->outq, PA_MSGOBJECT(u->module->core), PA_CORE_MESSAGE_UNLOAD_MODULE, u->module, 0, NULL, NULL);
+ }
+}
+
+static void on_sink_created(struct userdata *u) {
+ pa_proplist *proplist;
+ pa_buffer_attr bufferattr;
+ pa_usec_t requested_latency;
+ char *username = pa_get_user_name_malloc();
+ char *hostname = pa_get_host_name_malloc();
+ /* TODO: old tunnel put here the remote sink_name into stream name e.g. 'Null Output for lynxis at lazus' */
+ char *stream_name = pa_sprintf_malloc(_("Tunnel for %s@%s"), username, hostname);
+ pa_xfree(hostname);
+ pa_xfree(username);
+
+ pa_assert_io_context();
+
+ /* if we still don't have a sink, then sink creation failed, and we should
+ * kill this io thread */
+ if (!u->sink) {
+ pa_log_error("Could not create a sink.");
+ u->thread_mainloop_api->quit(u->thread_mainloop_api, TUNNEL_THREAD_FAILED_MAINLOOP);
+ return;
+ }
+
+ proplist = tunnel_new_proplist(u);
+ u->stream = pa_stream_new_with_proplist(u->context,
+ stream_name,
+ &u->sink->sample_spec,
+ &u->sink->channel_map,
+ proplist);
+ pa_proplist_free(proplist);
+ pa_xfree(stream_name);
+
+ if (!u->stream) {
+ pa_log_error("Could not create a stream.");
+ u->thread_mainloop_api->quit(u->thread_mainloop_api, TUNNEL_THREAD_FAILED_MAINLOOP);
+ return;
+ }
+
+ requested_latency = pa_sink_get_requested_latency_within_thread(u->sink);
+ if (requested_latency == (pa_usec_t) -1)
+ requested_latency = u->sink->thread_info.max_latency;
+
+ reset_bufferattr(&bufferattr);
+ bufferattr.tlength = pa_usec_to_bytes(requested_latency, &u->sink->sample_spec);
+
+ pa_log_debug("tlength requested at %lu.", (unsigned long) bufferattr.tlength);
+
+ pa_stream_set_state_callback(u->stream, stream_state_cb, u);
+ pa_stream_set_buffer_attr_callback(u->stream, stream_changed_buffer_attr_cb, u);
+ pa_stream_set_underflow_callback(u->stream, stream_underflow_callback, u);
+ pa_stream_set_overflow_callback(u->stream, stream_overflow_callback, u);
+ if (pa_stream_connect_playback(u->stream,
+ u->remote_sink_name,
+ &bufferattr,
+ PA_STREAM_INTERPOLATE_TIMING | PA_STREAM_DONT_MOVE | PA_STREAM_START_CORKED | PA_STREAM_AUTO_TIMING_UPDATE,
+ NULL,
+ NULL) < 0) {
+ pa_log_error("Could not connect stream.");
+ u->thread_mainloop_api->quit(u->thread_mainloop_api, TUNNEL_THREAD_FAILED_MAINLOOP);
+ }
+ u->connected = true;
+}
+
static void context_state_cb(pa_context *c, void *userdata) {
struct userdata *u = userdata;
pa_assert(u);
@@ -310,60 +417,18 @@ static void context_state_cb(pa_context *c, void *userdata) {
case PA_CONTEXT_AUTHORIZING:
case PA_CONTEXT_SETTING_NAME:
break;
- case PA_CONTEXT_READY: {
- pa_proplist *proplist;
- pa_buffer_attr bufferattr;
- pa_usec_t requested_latency;
- char *username = pa_get_user_name_malloc();
- char *hostname = pa_get_host_name_malloc();
- /* TODO: old tunnel put here the remote sink_name into stream name e.g. 'Null Output for lynxis at lazus' */
- char *stream_name = pa_sprintf_malloc(_("Tunnel for %s@%s"), username, hostname);
- pa_xfree(hostname);
- pa_xfree(username);
-
+ case PA_CONTEXT_READY:
+ /* now that we're connected, ask the control thread to create a sink for
+ * us, and wait for that to complete before proceeding, we'll
+ * receive TUNNEL_MESSAGE_SINK_CREATED in response when the sink is
+ * created (see sink_process_msg_cb()) */
pa_log_debug("Connection successful. Creating stream.");
pa_assert(!u->stream);
+ pa_assert(!u->sink);
- proplist = tunnel_new_proplist(u);
- u->stream = pa_stream_new_with_proplist(u->context,
- stream_name,
- &u->sink->sample_spec,
- &u->sink->channel_map,
- proplist);
- pa_proplist_free(proplist);
- pa_xfree(stream_name);
-
- if (!u->stream) {
- pa_log_error("Could not create a stream.");
- u->thread_mainloop_api->quit(u->thread_mainloop_api, TUNNEL_THREAD_FAILED_MAINLOOP);
- return;
- }
-
- requested_latency = pa_sink_get_requested_latency_within_thread(u->sink);
- if (requested_latency == (pa_usec_t) -1)
- requested_latency = u->sink->thread_info.max_latency;
-
- reset_bufferattr(&bufferattr);
- bufferattr.tlength = pa_usec_to_bytes(requested_latency, &u->sink->sample_spec);
-
- pa_log_debug("tlength requested at %lu.", (unsigned long) bufferattr.tlength);
-
- pa_stream_set_state_callback(u->stream, stream_state_cb, userdata);
- pa_stream_set_buffer_attr_callback(u->stream, stream_changed_buffer_attr_cb, userdata);
- pa_stream_set_underflow_callback(u->stream, stream_underflow_callback, userdata);
- pa_stream_set_overflow_callback(u->stream, stream_overflow_callback, userdata);
- if (pa_stream_connect_playback(u->stream,
- u->remote_sink_name,
- &bufferattr,
- PA_STREAM_INTERPOLATE_TIMING | PA_STREAM_DONT_MOVE | PA_STREAM_START_CORKED | PA_STREAM_AUTO_TIMING_UPDATE,
- NULL,
- NULL) < 0) {
- pa_log_error("Could not connect stream.");
- u->thread_mainloop_api->quit(u->thread_mainloop_api, TUNNEL_THREAD_FAILED_MAINLOOP);
- }
- u->connected = true;
+ 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);
break;
- }
case PA_CONTEXT_FAILED:
pa_log_debug("Context failed: %s.", pa_strerror(pa_context_errno(u->context)));
u->connected = false;
@@ -449,6 +514,9 @@ static int sink_process_msg_cb(pa_msgobject *o, int code, void *data, int64_t of
*((int64_t*) data) = remote_latency;
return 0;
}
+ case TUNNEL_MESSAGE_SINK_CREATED:
+ on_sink_created(u);
+ return 0;
}
return pa_sink_process_msg(o, code, data, offset, chunk);
}
@@ -487,15 +555,84 @@ static int sink_set_state_in_io_thread_cb(pa_sink *s, pa_sink_state_t new_state,
return 0;
}
-int pa__init(pa_module *m) {
+/* Creates a sink in the main thread.
+ *
+ * This method is called when we receive a message from the io thread that a
+ * connection has been established with the server. We defer creation of the
+ * sink until the connection is established, because we don't have a sink if
+ * the remote server isn't there.
+ */
+static void create_sink(struct userdata *u) {
+ pa_sink_new_data sink_data;
+
+ pa_assert_ctl_context();
+
+ /* Create sink */
+ pa_sink_new_data_init(&sink_data);
+ sink_data.driver = __FILE__;
+ sink_data.module = u->module;
+
+ pa_sink_new_data_set_name(&sink_data, u->sink_name);
+ pa_sink_new_data_set_sample_spec(&sink_data, &u->sample_spec);
+ pa_sink_new_data_set_channel_map(&sink_data, &u->channel_map);
+
+ pa_proplist_update(sink_data.proplist, PA_UPDATE_REPLACE, u->sink_proplist);
+
+ if (!(u->sink = pa_sink_new(u->module->core, &sink_data, PA_SINK_LATENCY | PA_SINK_DYNAMIC_LATENCY | PA_SINK_NETWORK))) {
+ pa_log("Failed to create sink.");
+ goto finish;
+ }
+
+ u->sink->userdata = u;
+ u->sink->parent.process_msg = sink_process_msg_cb;
+ u->sink->set_state_in_io_thread = sink_set_state_in_io_thread_cb;
+ u->sink->update_requested_latency = sink_update_requested_latency_cb;
+ pa_sink_set_latency_range(u->sink, 0, MAX_LATENCY_USEC);
+
+ /* set thread message queue */
+ pa_sink_set_asyncmsgq(u->sink, u->thread_mq->inq);
+ pa_sink_set_rtpoll(u->sink, u->rtpoll);
+
+ pa_sink_put(u->sink);
+
+finish:
+ pa_sink_new_data_done(&sink_data);
+
+ /* tell any interested io threads that the sink they asked for has now been
+ * created (even if we failed, we still notify the thread, so they can
+ * either handle or kill the thread, rather than deadlock waiting for a
+ * message that will never come */
+ pa_asyncmsgq_send(u->sink->asyncmsgq, PA_MSGOBJECT(u->sink), TUNNEL_MESSAGE_SINK_CREATED, u, 0, NULL);
+}
+
+/* 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) {
+ case TUNNEL_MESSAGE_CREATE_SINK_REQUEST:
+ create_sink(u);
+ break;
+ case TUNNEL_MESSAGE_MAYBE_RESTART:
+ maybe_restart(u);
+ break;
+ }
+
+ return 0;
+}
+
+static int do_init(pa_module *m) {
struct userdata *u = NULL;
pa_modargs *ma = NULL;
- pa_sink_new_data sink_data;
- pa_sample_spec ss;
- pa_channel_map map;
const char *remote_server = NULL;
- const char *sink_name = NULL;
char *default_sink_name = NULL;
+ uint32_t reconnect_interval_ms = 0;
pa_assert(m);
@@ -504,9 +641,13 @@ int pa__init(pa_module *m) {
goto fail;
}
- 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 = pa_xnew0(struct userdata, 1);
+ u->module = m;
+ m->userdata = u;
+
+ 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 or channel map");
goto fail;
}
@@ -517,9 +658,6 @@ int pa__init(pa_module *m) {
goto fail;
}
- u = pa_xnew0(struct userdata, 1);
- u->module = m;
- m->userdata = u;
u->remote_server = pa_xstrdup(remote_server);
u->thread_mainloop = pa_mainloop_new();
if (u->thread_mainloop == NULL) {
@@ -537,6 +675,9 @@ int pa__init(pa_module *m) {
goto fail;
}
+ u->msg = pa_msgobject_new(tunnel_msg);
+ u->msg->parent.process_msg = tunnel_process_msg;
+
/* The rtpoll created here is never run. It is only necessary to avoid crashes
* when module-tunnel-sink-new is used together with module-loopback or
* module-combine-sink. Both modules base their asyncmsq on the rtpoll provided
@@ -546,53 +687,30 @@ int pa__init(pa_module *m) {
* with module-tunnel-sink-new. */
u->rtpoll = pa_rtpoll_new();
- /* Create sink */
- pa_sink_new_data_init(&sink_data);
- sink_data.driver = __FILE__;
- sink_data.module = m;
-
default_sink_name = pa_sprintf_malloc("tunnel-sink-new.%s", remote_server);
- sink_name = pa_modargs_get_value(ma, "sink_name", default_sink_name);
-
- pa_sink_new_data_set_name(&sink_data, sink_name);
- pa_sink_new_data_set_sample_spec(&sink_data, &ss);
- pa_sink_new_data_set_channel_map(&sink_data, &map);
+ u->sink_name = pa_xstrdup(pa_modargs_get_value(ma, "sink_name", default_sink_name));
- pa_proplist_sets(sink_data.proplist, PA_PROP_DEVICE_CLASS, "sound");
- pa_proplist_setf(sink_data.proplist,
+ u->sink_proplist = pa_proplist_new();
+ pa_proplist_sets(u->sink_proplist, PA_PROP_DEVICE_CLASS, "sound");
+ pa_proplist_setf(u->sink_proplist,
PA_PROP_DEVICE_DESCRIPTION,
_("Tunnel to %s/%s"),
remote_server,
pa_strempty(u->remote_sink_name));
- if (pa_modargs_get_proplist(ma, "sink_properties", sink_data.proplist, PA_UPDATE_REPLACE) < 0) {
+ if (pa_modargs_get_proplist(ma, "sink_properties", u->sink_proplist, PA_UPDATE_REPLACE) < 0) {
pa_log("Invalid properties");
- pa_sink_new_data_done(&sink_data);
- goto fail;
- }
- if (!(u->sink = pa_sink_new(m->core, &sink_data, PA_SINK_LATENCY | PA_SINK_DYNAMIC_LATENCY | PA_SINK_NETWORK))) {
- pa_log("Failed to create sink.");
- pa_sink_new_data_done(&sink_data);
goto fail;
}
- pa_sink_new_data_done(&sink_data);
- u->sink->userdata = u;
- u->sink->parent.process_msg = sink_process_msg_cb;
- u->sink->set_state_in_io_thread = sink_set_state_in_io_thread_cb;
- u->sink->update_requested_latency = sink_update_requested_latency_cb;
- pa_sink_set_latency_range(u->sink, 0, MAX_LATENCY_USEC);
-
- /* set thread message queue */
- pa_sink_set_asyncmsgq(u->sink, u->thread_mq->inq);
- pa_sink_set_rtpoll(u->sink, u->rtpoll);
+ pa_modargs_get_value_u32(ma, "reconnect_interval_ms", &reconnect_interval_ms);
+ u->reconnect_interval_us = reconnect_interval_ms * PA_USEC_PER_MSEC;
if (!(u->thread = pa_thread_new("tunnel-sink", thread_func, u))) {
pa_log("Failed to create thread.");
goto fail;
}
- pa_sink_put(u->sink);
pa_modargs_free(ma);
pa_xfree(default_sink_name);
@@ -605,19 +723,19 @@ fail:
if (default_sink_name)
pa_xfree(default_sink_name);
- pa__done(m);
-
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;
+
if (u->sink)
pa_sink_unlink(u->sink);
@@ -649,5 +767,34 @@ void pa__done(pa_module *m) {
if (u->rtpoll)
pa_rtpoll_free(u->rtpoll);
+ if (u->sink_proplist)
+ pa_proplist_free(u->sink_proplist);
+
+ if (u->sink_name)
+ pa_xfree(u->sink_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);
}
=====================================
src/modules/module-tunnel-source-new.c
=====================================
@@ -21,6 +21,8 @@
#include <config.h>
#endif
+#include "restart-module.h"
+
#include <pulse/context.h>
#include <pulse/timeval.h>
#include <pulse/xmalloc.h>
@@ -50,6 +52,7 @@ PA_MODULE_USAGE(
"source=<name of the remote source> "
"source_name=<name for the local source> "
"source_properties=<properties for the local source> "
+ "reconnect_interval_ms=<interval to try reconnects, 0 or omitted if disabled> "
"format=<sample format> "
"channels=<number of channels> "
"rate=<sample rate> "
@@ -59,11 +62,29 @@ PA_MODULE_USAGE(
#define TUNNEL_THREAD_FAILED_MAINLOOP 1
+static int do_init(pa_module *m);
+static void do_done(pa_module *m);
static void stream_state_cb(pa_stream *stream, void *userdata);
static void stream_read_cb(pa_stream *s, size_t length, void *userdata);
static void context_state_cb(pa_context *c, void *userdata);
static void source_update_requested_latency_cb(pa_source *s);
+struct tunnel_msg {
+ pa_msgobject parent;
+};
+
+typedef struct tunnel_msg tunnel_msg;
+PA_DEFINE_PRIVATE_CLASS(tunnel_msg, pa_msgobject);
+
+enum {
+ TUNNEL_MESSAGE_CREATE_SOURCE_REQUEST,
+ TUNNEL_MESSAGE_MAYBE_RESTART,
+};
+
+enum {
+ TUNNEL_MESSAGE_SOURCE_CREATED = PA_SOURCE_MESSAGE_MAX,
+};
+
struct userdata {
pa_module *module;
pa_source *source;
@@ -78,11 +99,21 @@ struct userdata {
bool update_stream_bufferattr_after_connect;
bool connected;
+ bool shutting_down;
bool new_data;
char *cookie_file;
char *remote_server;
char *remote_source_name;
+ char *source_name;
+
+ pa_proplist *source_proplist;
+ pa_sample_spec sample_spec;
+ pa_channel_map channel_map;
+
+ tunnel_msg *msg;
+
+ pa_usec_t reconnect_interval_us;
};
static const char* const valid_modargs[] = {
@@ -95,7 +126,7 @@ static const char* const valid_modargs[] = {
"rate",
"channel_map",
"cookie",
- /* "reconnect", reconnect if server comes back again - unimplemented */
+ "reconnect_interval_ms",
NULL,
};
@@ -236,7 +267,10 @@ static void thread_func(void *userdata) {
read_new_samples(u);
}
fail:
- pa_asyncmsgq_post(u->thread_mq->outq, PA_MSGOBJECT(u->module->core), PA_CORE_MESSAGE_UNLOAD_MODULE, u->module, 0, NULL, NULL);
+ /* send a message to the ctl thread to ask it to either terminate us, or
+ * restart us, but either way this thread will exit, so then wait for the
+ * shutdown message */
+ 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:
@@ -285,6 +319,74 @@ static void stream_state_cb(pa_stream *stream, void *userdata) {
}
}
+/* 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 maybe_restart(struct userdata *u) {
+ if (u->reconnect_interval_us > 0) {
+ pa_restart_module_reinit(u->module, do_init, do_done, u->reconnect_interval_us);
+ } else {
+ /* exit the module */
+ pa_asyncmsgq_post(u->thread_mq->outq, PA_MSGOBJECT(u->module->core), PA_CORE_MESSAGE_UNLOAD_MODULE, u->module, 0, NULL, NULL);
+ }
+}
+
+static void on_source_created(struct userdata *u) {
+ pa_proplist *proplist;
+ pa_buffer_attr bufferattr;
+ pa_usec_t requested_latency;
+ char *username = pa_get_user_name_malloc();
+ char *hostname = pa_get_host_name_malloc();
+ /* TODO: old tunnel put here the remote source_name into stream name e.g. 'Null Output for lynxis at lazus' */
+ char *stream_name = pa_sprintf_malloc(_("Tunnel for %s@%s"), username, hostname);
+ pa_xfree(username);
+ pa_xfree(hostname);
+
+ pa_assert_io_context();
+
+ /* if we still don't have a source, then source creation failed, and we
+ * should kill this io thread */
+ if (!u->source) {
+ pa_log_error("Could not create a source.");
+ u->thread_mainloop_api->quit(u->thread_mainloop_api, TUNNEL_THREAD_FAILED_MAINLOOP);
+ return;
+ }
+
+ proplist = tunnel_new_proplist(u);
+ u->stream = pa_stream_new_with_proplist(u->context,
+ stream_name,
+ &u->source->sample_spec,
+ &u->source->channel_map,
+ proplist);
+ pa_proplist_free(proplist);
+ pa_xfree(stream_name);
+
+ if (!u->stream) {
+ pa_log_error("Could not create a stream: %s", pa_strerror(pa_context_errno(u->context)));
+ u->thread_mainloop_api->quit(u->thread_mainloop_api, TUNNEL_THREAD_FAILED_MAINLOOP);
+ return;
+ }
+
+ requested_latency = pa_source_get_requested_latency_within_thread(u->source);
+ if (requested_latency == (uint32_t) -1)
+ requested_latency = u->source->thread_info.max_latency;
+
+ reset_bufferattr(&bufferattr);
+ bufferattr.fragsize = pa_usec_to_bytes(requested_latency, &u->source->sample_spec);
+
+ pa_stream_set_state_callback(u->stream, stream_state_cb, u);
+ pa_stream_set_read_callback(u->stream, stream_read_cb, u);
+ if (pa_stream_connect_record(u->stream,
+ u->remote_source_name,
+ &bufferattr,
+ PA_STREAM_INTERPOLATE_TIMING|PA_STREAM_DONT_MOVE|PA_STREAM_AUTO_TIMING_UPDATE|PA_STREAM_START_CORKED) < 0) {
+ pa_log_debug("Could not create stream: %s", pa_strerror(pa_context_errno(u->context)));
+ u->thread_mainloop_api->quit(u->thread_mainloop_api, TUNNEL_THREAD_FAILED_MAINLOOP);
+ }
+ u->connected = true;
+}
+
static void context_state_cb(pa_context *c, void *userdata) {
struct userdata *u = userdata;
pa_assert(u);
@@ -295,54 +397,14 @@ static void context_state_cb(pa_context *c, void *userdata) {
case PA_CONTEXT_AUTHORIZING:
case PA_CONTEXT_SETTING_NAME:
break;
- case PA_CONTEXT_READY: {
- pa_proplist *proplist;
- pa_buffer_attr bufferattr;
- pa_usec_t requested_latency;
- char *username = pa_get_user_name_malloc();
- char *hostname = pa_get_host_name_malloc();
- /* TODO: old tunnel put here the remote source_name into stream name e.g. 'Null Output for lynxis at lazus' */
- char *stream_name = pa_sprintf_malloc(_("Tunnel for %s@%s"), username, hostname);
- pa_xfree(username);
- pa_xfree(hostname);
-
+ case PA_CONTEXT_READY:
pa_log_debug("Connection successful. Creating stream.");
pa_assert(!u->stream);
+ pa_assert(!u->source);
- proplist = tunnel_new_proplist(u);
- u->stream = pa_stream_new_with_proplist(u->context,
- stream_name,
- &u->source->sample_spec,
- &u->source->channel_map,
- proplist);
- pa_proplist_free(proplist);
- pa_xfree(stream_name);
-
- if (!u->stream) {
- pa_log_error("Could not create a stream: %s", pa_strerror(pa_context_errno(u->context)));
- u->thread_mainloop_api->quit(u->thread_mainloop_api, TUNNEL_THREAD_FAILED_MAINLOOP);
- return;
- }
-
- requested_latency = pa_source_get_requested_latency_within_thread(u->source);
- if (requested_latency == (uint32_t) -1)
- requested_latency = u->source->thread_info.max_latency;
-
- reset_bufferattr(&bufferattr);
- bufferattr.fragsize = pa_usec_to_bytes(requested_latency, &u->source->sample_spec);
-
- pa_stream_set_state_callback(u->stream, stream_state_cb, userdata);
- pa_stream_set_read_callback(u->stream, stream_read_cb, userdata);
- if (pa_stream_connect_record(u->stream,
- u->remote_source_name,
- &bufferattr,
- PA_STREAM_INTERPOLATE_TIMING|PA_STREAM_DONT_MOVE|PA_STREAM_AUTO_TIMING_UPDATE|PA_STREAM_START_CORKED) < 0) {
- pa_log_debug("Could not create stream: %s", pa_strerror(pa_context_errno(u->context)));
- u->thread_mainloop_api->quit(u->thread_mainloop_api, TUNNEL_THREAD_FAILED_MAINLOOP);
- }
- u->connected = true;
+ 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);
break;
- }
case PA_CONTEXT_FAILED:
pa_log_debug("Context failed with err %s.", pa_strerror(pa_context_errno(u->context)));
u->connected = false;
@@ -428,6 +490,9 @@ static int source_process_msg_cb(pa_msgobject *o, int code, void *data, int64_t
return 0;
}
+ case TUNNEL_MESSAGE_SOURCE_CREATED:
+ on_source_created(u);
+ return 0;
}
return pa_source_process_msg(o, code, data, offset, chunk);
}
@@ -466,15 +531,82 @@ static int source_set_state_in_io_thread_cb(pa_source *s, pa_source_state_t new_
return 0;
}
-int pa__init(pa_module *m) {
+/* Creates a source in the main thread.
+ *
+ * This method is called when we receive a message from the io thread that a
+ * connection has been established with the server. We defer creation of the
+ * source until the connection is established, because we don't have a source
+ * if the remote server isn't there.
+ */
+static void create_source(struct userdata *u) {
+ pa_source_new_data source_data;
+
+ pa_assert_ctl_context();
+
+ /* Create source */
+ pa_source_new_data_init(&source_data);
+ source_data.driver = __FILE__;
+ source_data.module = u->module;
+
+ pa_source_new_data_set_name(&source_data, u->source_name);
+ pa_source_new_data_set_sample_spec(&source_data, &u->sample_spec);
+ pa_source_new_data_set_channel_map(&source_data, &u->channel_map);
+
+ pa_proplist_update(source_data.proplist, PA_UPDATE_REPLACE, u->source_proplist);
+
+ if (!(u->source = pa_source_new(u->module->core, &source_data, PA_SOURCE_LATENCY | PA_SOURCE_DYNAMIC_LATENCY | PA_SOURCE_NETWORK))) {
+ pa_log("Failed to create source.");
+ goto finish;
+ }
+
+ u->source->userdata = u;
+ u->source->parent.process_msg = source_process_msg_cb;
+ u->source->set_state_in_io_thread = source_set_state_in_io_thread_cb;
+ u->source->update_requested_latency = source_update_requested_latency_cb;
+
+ pa_source_set_asyncmsgq(u->source, u->thread_mq->inq);
+ pa_source_set_rtpoll(u->source, u->rtpoll);
+
+ pa_source_put(u->source);
+
+finish:
+ pa_source_new_data_done(&source_data);
+
+ /* tell any interested io threads that the sink they asked for has now been
+ * created (even if we failed, we still notify the thread, so they can
+ * either handle or kill the thread, rather than deadlock waiting for a
+ * message that will never come */
+ pa_asyncmsgq_send(u->source->asyncmsgq, PA_MSGOBJECT(u->source), TUNNEL_MESSAGE_SOURCE_CREATED, u, 0, NULL);
+}
+
+/* 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) {
+ case TUNNEL_MESSAGE_CREATE_SOURCE_REQUEST:
+ create_source(u);
+ break;
+ case TUNNEL_MESSAGE_MAYBE_RESTART:
+ maybe_restart(u);
+ break;
+ }
+
+ return 0;
+}
+
+static int do_init(pa_module *m) {
struct userdata *u = NULL;
pa_modargs *ma = NULL;
- pa_source_new_data source_data;
- pa_sample_spec ss;
- pa_channel_map map;
const char *remote_server = NULL;
- const char *source_name = NULL;
char *default_source_name = NULL;
+ uint32_t reconnect_interval_ms = 0;
pa_assert(m);
@@ -483,9 +615,13 @@ int pa__init(pa_module *m) {
goto fail;
}
- 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 = pa_xnew0(struct userdata, 1);
+ u->module = m;
+ m->userdata = u;
+
+ 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 or channel map");
goto fail;
}
@@ -496,9 +632,6 @@ int pa__init(pa_module *m) {
goto fail;
}
- u = pa_xnew0(struct userdata, 1);
- u->module = m;
- m->userdata = u;
u->remote_server = pa_xstrdup(remote_server);
u->thread_mainloop = pa_mainloop_new();
if (u->thread_mainloop == NULL) {
@@ -516,57 +649,39 @@ int pa__init(pa_module *m) {
goto fail;
}
+ u->msg = pa_msgobject_new(tunnel_msg);
+ u->msg->parent.process_msg = tunnel_process_msg;
+
/* The rtpoll created here is never run. It is only necessary to avoid crashes
* when module-tunnel-source-new is used together with module-loopback.
* module-loopback bases the asyncmsq on the rtpoll provided by the source and
* only works because it calls pa_asyncmsq_process_one(). */
u->rtpoll = pa_rtpoll_new();
- /* Create source */
- pa_source_new_data_init(&source_data);
- source_data.driver = __FILE__;
- source_data.module = m;
-
default_source_name = pa_sprintf_malloc("tunnel-source-new.%s", remote_server);
- source_name = pa_modargs_get_value(ma, "source_name", default_source_name);
-
- pa_source_new_data_set_name(&source_data, source_name);
- pa_source_new_data_set_sample_spec(&source_data, &ss);
- pa_source_new_data_set_channel_map(&source_data, &map);
+ u->source_name = pa_xstrdup(pa_modargs_get_value(ma, "source_name", default_source_name));
- pa_proplist_sets(source_data.proplist, PA_PROP_DEVICE_CLASS, "sound");
- pa_proplist_setf(source_data.proplist,
+ u->source_proplist = pa_proplist_new();
+ pa_proplist_sets(u->source_proplist, PA_PROP_DEVICE_CLASS, "sound");
+ pa_proplist_setf(u->source_proplist,
PA_PROP_DEVICE_DESCRIPTION,
_("Tunnel to %s/%s"),
remote_server,
pa_strempty(u->remote_source_name));
- if (pa_modargs_get_proplist(ma, "source_properties", source_data.proplist, PA_UPDATE_REPLACE) < 0) {
+ if (pa_modargs_get_proplist(ma, "source_properties", u->source_proplist, PA_UPDATE_REPLACE) < 0) {
pa_log("Invalid properties");
- pa_source_new_data_done(&source_data);
- goto fail;
- }
- if (!(u->source = pa_source_new(m->core, &source_data, PA_SOURCE_LATENCY | PA_SOURCE_DYNAMIC_LATENCY | PA_SOURCE_NETWORK))) {
- pa_log("Failed to create source.");
- pa_source_new_data_done(&source_data);
goto fail;
}
- pa_source_new_data_done(&source_data);
- u->source->userdata = u;
- u->source->parent.process_msg = source_process_msg_cb;
- u->source->set_state_in_io_thread = source_set_state_in_io_thread_cb;
- u->source->update_requested_latency = source_update_requested_latency_cb;
-
- pa_source_set_asyncmsgq(u->source, u->thread_mq->inq);
- pa_source_set_rtpoll(u->source, u->rtpoll);
+ pa_modargs_get_value_u32(ma, "reconnect_interval_ms", &reconnect_interval_ms);
+ u->reconnect_interval_us = reconnect_interval_ms * PA_USEC_PER_MSEC;
if (!(u->thread = pa_thread_new("tunnel-source", thread_func, u))) {
pa_log("Failed to create thread.");
goto fail;
}
- pa_source_put(u->source);
pa_modargs_free(ma);
pa_xfree(default_source_name);
@@ -579,19 +694,19 @@ fail:
if (default_source_name)
pa_xfree(default_source_name);
- pa__done(m);
-
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;
+
if (u->source)
pa_source_unlink(u->source);
@@ -623,5 +738,34 @@ void pa__done(pa_module *m) {
if (u->rtpoll)
pa_rtpoll_free(u->rtpoll);
+ if (u->source_proplist)
+ pa_proplist_free(u->source_proplist);
+
+ if (u->source_name)
+ pa_xfree(u->source_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);
}
=====================================
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;
}
}
@@ -612,6 +671,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;
}
@@ -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);
}
=====================================
src/modules/restart-module.c
=====================================
@@ -0,0 +1,88 @@
+/***
+ This file is part of PulseAudio.
+
+ Copyright 2022 Craig Howard
+
+ PulseAudio is free software; you can redistribute it and/or modify
+ it under the terms of the GNU Lesser General Public License as published
+ by the Free Software Foundation; either version 2.1 of the License,
+ or (at your option) any later version.
+
+ PulseAudio is distributed in the hope that it will be useful, but
+ WITHOUT ANY WARRANTY; without even the implied warranty of
+ MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
+ General Public License for more details.
+
+ You should have received a copy of the GNU Lesser General Public License
+ along with PulseAudio; if not, see <http://www.gnu.org/licenses/>.
+***/
+
+#ifdef HAVE_CONFIG_H
+#include <config.h>
+#endif
+
+#include "restart-module.h"
+
+#include <pulse/timeval.h>
+#include <pulse/xmalloc.h>
+#include <pulse/mainloop.h>
+
+#include <pulsecore/core.h>
+#include <pulsecore/thread-mq.h>
+
+struct reinit_data {
+ init_cb do_init;
+ done_cb do_done;
+
+ pa_usec_t restart_usec;
+ pa_module *module;
+};
+
+static void call_init(pa_mainloop_api *mainloop, pa_time_event *e, const struct timeval *tv, void *userdata) {
+ struct reinit_data *rd = userdata;
+ int ret;
+
+ /* now that restart_usec has elapsed, we call do_init to restart the module */
+ ret = rd->do_init(rd->module);
+
+ /* if the init failed, we got here because the caller wanted to restart, so
+ * setup another restart */
+ if (ret < 0)
+ pa_restart_module_reinit(rd->module, rd->do_init, rd->do_done, rd->restart_usec);
+
+ pa_xfree(rd);
+}
+
+static void do_reinit(pa_mainloop_api *mainloop, void *userdata) {
+ struct reinit_data *rd = userdata;
+ struct timeval tv;
+
+ pa_assert_ctl_context();
+
+ /* call do_done on the module, which will effectively tear it down; all
+ * that remains is the pa_module */
+ rd->do_done(rd->module);
+
+ /* after restart_usec, call do_init to restart the module */
+ pa_gettimeofday(&tv);
+ pa_timeval_add(&tv, rd->restart_usec);
+ mainloop->time_new(mainloop, &tv, call_init, rd);
+}
+
+void pa_restart_module_reinit(pa_module *m, init_cb do_init, done_cb do_done, pa_usec_t restart_usec) {
+ struct reinit_data *rd;
+
+ pa_assert_ctl_context();
+
+ pa_log_info("Starting reinit for %s", m->name);
+
+ rd = pa_xnew0(struct reinit_data, 1);
+ rd->do_init = do_init;
+ rd->do_done = do_done;
+ rd->restart_usec = restart_usec;
+ rd->module = m;
+
+ /* defer actually doing a reinit, so that we can safely exit whatever call
+ * chain we're in before we effectively reinit the module */
+ pa_mainloop_api_once(m->core->mainloop, do_reinit, rd);
+}
=====================================
src/modules/restart-module.h
=====================================
@@ -0,0 +1,41 @@
+/***
+ This file is part of PulseAudio.
+
+ Copyright 2022 Craig Howard
+
+ PulseAudio is free software; you can redistribute it and/or modify
+ it under the terms of the GNU Lesser General Public License as published
+ by the Free Software Foundation; either version 2.1 of the License,
+ or (at your option) any later version.
+
+ PulseAudio is distributed in the hope that it will be useful, but
+ WITHOUT ANY WARRANTY; without even the implied warranty of
+ MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
+ General Public License for more details.
+
+ You should have received a copy of the GNU Lesser General Public License
+ along with PulseAudio; if not, see <http://www.gnu.org/licenses/>.
+***/
+
+#ifndef RESTART_MODULE_H
+#define RESTART_MODULE_H
+
+#ifdef __cplusplus
+extern "C" {
+#endif
+
+#include <pulse/timeval.h>
+
+#include <pulsecore/core.h>
+#include <pulsecore/thread-mq.h>
+
+typedef int (*init_cb)(pa_module *m);
+typedef void (*done_cb)(pa_module *m);
+
+void pa_restart_module_reinit(pa_module *m, init_cb do_init, done_cb do_done, pa_usec_t restart_usec);
+
+#ifdef __cplusplus
+}
+#endif
+
+#endif
View it on GitLab: https://gitlab.freedesktop.org/pulseaudio/pulseaudio/-/compare/c3c84a3691e700157ccd3c75fdae53c6eec89775...17eb178f17a0ebb07d2355615a1f6ba0d074399d
--
View it on GitLab: https://gitlab.freedesktop.org/pulseaudio/pulseaudio/-/compare/c3c84a3691e700157ccd3c75fdae53c6eec89775...17eb178f17a0ebb07d2355615a1f6ba0d074399d
You're receiving this email because of your account on gitlab.freedesktop.org.
-------------- next part --------------
An HTML attachment was scrubbed...
URL: <https://lists.freedesktop.org/archives/pulseaudio-commits/attachments/20220516/b0936576/attachment-0001.htm>
More information about the pulseaudio-commits
mailing list