Merge pull request #59 from matt335672/module_params

Module load and unload support

2 files changed, 184 insertions, 67 deletions

diff --git a/src/module-xrdp-sink.c b/src/module-xrdp-sink.c
index 91c56d9..824c7bb 100644
--- a/src/module-xrdp-sink.c
+++ b/src/module-xrdp-sink.c
@@ -81,11 +81,21 @@ PA_MODULE_USAGE(
         "format=<sample format> "
         "rate=<sample rate> "
         "channels=<number of channels> "
-        "channel_map=<channel map>");
+        "channel_map=<channel map> "
+        "xrdp_socket_path=<path to XRDP sockets> "
+        "xrdp_pulse_sink_socket=<name of sink socket>");
 
 #define DEFAULT_SINK_NAME "xrdp-sink"
 #define BLOCK_USEC 30000
 //#define BLOCK_USEC (PA_USEC_PER_SEC * 2)
+#define UNUSED_VAR(x) ((void) (x))
+
+/* support for the set_state_in_io_thread callback was added in 11.99.1 */
+#if defined(PA_CHECK_VERSION) && PA_CHECK_VERSION(11, 99, 1)
+#define USE_SET_STATE_IN_IO_THREAD_CB
+#else
+#undef USE_SET_STATE_IN_IO_THREAD_CB
+#endif
 
 struct userdata {
     pa_core *core;
@@ -102,10 +112,9 @@ struct userdata {
     pa_usec_t last_send_time;
 
     int fd; /* unix domain socket connection to xrdp chansrv */
-    int display_num;
     int skip_bytes;
-    int got_max_latency;
 
+    char *sink_socket;
 };
 
 static const char* const valid_modargs[] = {
@@ -115,6 +124,8 @@ static const char* const valid_modargs[] = {
     "rate",
     "channels",
     "channel_map",
+    "xrdp_socket_path",
+    "xrdp_pulse_sink_socket",
     NULL
 };
 
@@ -127,27 +138,30 @@ static int sink_process_msg(pa_msgobject *o, int code, void *data,
     pa_usec_t now;
     long lat;
 
-    pa_log_debug("sink_process_msg: code %d", code);
-
     switch (code) {
 
-        case PA_SINK_MESSAGE_SET_VOLUME: /* 3 */
+        case PA_SINK_MESSAGE_SET_VOLUME:
+            pa_log_debug("sink_process_msg: PA_SINK_MESSAGE_SET_VOLUME");
             break;
 
-        case PA_SINK_MESSAGE_SET_MUTE: /* 6 */
+        case PA_SINK_MESSAGE_SET_MUTE:
+            pa_log_debug("sink_process_msg: PA_SINK_MESSAGE_SET_MUTE");
             break;
 
-        case PA_SINK_MESSAGE_GET_LATENCY: /* 7 */
+        case PA_SINK_MESSAGE_GET_LATENCY:
+            pa_log_debug("sink_process_msg: PA_SINK_MESSAGE_GET_LATENCY");
             now = pa_rtclock_now();
             lat = u->timestamp > now ? u->timestamp - now : 0ULL;
             pa_log_debug("sink_process_msg: lat %ld", lat);
             *((pa_usec_t*) data) = lat;
             return 0;
 
-        case PA_SINK_MESSAGE_GET_REQUESTED_LATENCY: /* 8 */
+        case PA_SINK_MESSAGE_GET_REQUESTED_LATENCY:
+            pa_log_debug("sink_process_msg: PA_SINK_MESSAGE_GET_REQUESTED_LATENCY");
             break;
 
-        case PA_SINK_MESSAGE_SET_STATE: /* 9 */
+        case PA_SINK_MESSAGE_SET_STATE:
+            pa_log_debug("sink_process_msg: PA_SINK_MESSAGE_SET_STATE");
             if (PA_PTR_TO_UINT(data) == PA_SINK_RUNNING) /* 0 */ {
                 pa_log("sink_process_msg: running");
 
@@ -158,11 +172,40 @@ static int sink_process_msg(pa_msgobject *o, int code, void *data,
             }
             break;
 
+        default:
+            pa_log_debug("sink_process_msg: code %d", code);
+
     }
 
     return pa_sink_process_msg(o, code, data, offset, chunk);
 }
 
+#ifdef USE_SET_STATE_IN_IO_THREAD_CB
+/* Called from the IO thread. */
+static int sink_set_state_in_io_thread_cb(pa_sink *s,
+                                          pa_sink_state_t new_state,
+                                          pa_suspend_cause_t new_suspend_cause)
+{
+    struct userdata *u;
+
+    UNUSED_VAR(new_suspend_cause);
+
+    pa_assert(s);
+    pa_assert_se(u = s->userdata);
+
+    if (s->thread_info.state == PA_SINK_SUSPENDED || s->thread_info.state == PA_SINK_INIT)
+    {
+        if (PA_SINK_IS_OPENED(new_state))
+        {
+            pa_log_debug("sink_set_state_in_io_thread_cb: set timestamp");
+            u->timestamp = pa_rtclock_now();
+        }
+    }
+
+    return 0;
+}
+#endif /* USE_SET_STATE_IN_IO_THREAD_CB */
+
 static void sink_update_requested_latency_cb(pa_sink *s) {
     struct userdata *u;
     size_t nbytes;
@@ -176,15 +219,52 @@ static void sink_update_requested_latency_cb(pa_sink *s) {
         u->block_usec = s->thread_info.max_latency;
     }
     nbytes = pa_usec_to_bytes(u->block_usec, &s->sample_spec);
+    pa_sink_set_max_rewind_within_thread(s, nbytes);
     pa_sink_set_max_request_within_thread(s, nbytes);
 }
 
+static void process_rewind(struct userdata *u, pa_usec_t now) {
+    size_t rewind_nbytes, in_buffer;
+    pa_usec_t delay;
+
+    pa_assert(u);
+
+    rewind_nbytes = u->sink->thread_info.rewind_nbytes;
+
+    if (!PA_SINK_IS_OPENED(u->sink->thread_info.state) || rewind_nbytes <= 0)
+        goto do_nothing;
+
+    pa_log_debug("Requested to rewind %lu bytes.", (unsigned long) rewind_nbytes);
+
+    if (u->timestamp <= now)
+        goto do_nothing;
+
+    delay = u->timestamp - now;
+    in_buffer = pa_usec_to_bytes(delay, &u->sink->sample_spec);
+
+    if (in_buffer <= 0)
+        goto do_nothing;
+
+    if (rewind_nbytes > in_buffer)
+        rewind_nbytes = in_buffer;
+
+    pa_sink_process_rewind(u->sink, rewind_nbytes);
+    u->timestamp -= pa_bytes_to_usec(rewind_nbytes, &u->sink->sample_spec);
+
+    pa_log_debug("Rewound %lu bytes.", (unsigned long) rewind_nbytes);
+    return;
+
+do_nothing:
+
+    pa_sink_process_rewind(u->sink, 0);
+}
+
 struct header {
     int code;
     int bytes;
 };
 
-static int get_display_num_from_display(char *display_text) {
+static int get_display_num_from_display(const char *display_text) {
     int index;
     int mode;
     int host_index;
@@ -248,8 +328,6 @@ static int lsend(int fd, char *data, int bytes) {
 
 static int data_send(struct userdata *u, pa_memchunk *chunk) {
     char *data;
-    char *socket_dir;
-    char *sink_socket;
     int bytes;
     int sent;
     int fd;
@@ -265,26 +343,7 @@ static int data_send(struct userdata *u, pa_memchunk *chunk) {
         fd = socket(PF_LOCAL, SOCK_STREAM, 0);
         memset(&s, 0, sizeof(s));
         s.sun_family = AF_UNIX;
-        bytes = sizeof(s.sun_path) - 1;
-        socket_dir = getenv("XRDP_SOCKET_PATH");
-        if (socket_dir == NULL || socket_dir[0] == '\0')
-        {
-            socket_dir = "/tmp/.xrdp";
-        }
-        sink_socket = getenv("XRDP_PULSE_SINK_SOCKET");
-        if (sink_socket == NULL || sink_socket[0] == '\0')
-        {
-            pa_log_debug("Could not obtain sink_socket from environment.");
-            /* usually it doesn't reach here. if the socket name is not given
-               via environment variable, use hardcoded name as fallback */
-            snprintf(s.sun_path, bytes,
-                    "%s/xrdp_chansrv_audio_out_socket_%d", socket_dir, u->display_num);
-        }
-        else
-        {
-            pa_log_debug("Obtained sink_socket from environment.");
-            snprintf(s.sun_path, bytes, "%s/%s", socket_dir, sink_socket);
-        }
+        pa_strlcpy(s.sun_path, u->sink_socket, sizeof(s.sun_path));
         pa_log_debug("trying to connect to %s", s.sun_path);
 
         if (connect(fd, (struct sockaddr *)&s,
@@ -353,9 +412,6 @@ static void process_render(struct userdata *u, pa_usec_t now) {
     int request_bytes;
 
     pa_assert(u);
-    if (u->got_max_latency) {
-        return;
-    }
     pa_log_debug("process_render: u->block_usec %llu", (unsigned long long) u->block_usec);
     while (u->timestamp < now + u->block_usec) {
         request_bytes = u->sink->thread_info.max_request;
@@ -389,7 +445,7 @@ static void thread_func(void *userdata) {
             now = pa_rtclock_now();
         }
         if (PA_UNLIKELY(u->sink->thread_info.rewind_requested)) {
-            pa_sink_process_rewind(u->sink, 0);
+            process_rewind(u, now);
         }
         /* Render some data and write it to the socket */
         if (PA_SINK_IS_OPENED(u->sink->thread_info.state)) {
@@ -426,6 +482,35 @@ finish:
     pa_log_debug("Thread shutting down");
 }
 
+static void set_sink_socket(pa_modargs *ma, struct userdata *u) {
+    const char *socket_dir;
+    const char *socket_name;
+    char default_socket_name[64];
+    size_t nbytes;
+
+    socket_dir = pa_modargs_get_value(ma, "xrdp_socket_path",
+                                      getenv("XRDP_SOCKET_PATH"));
+    if (socket_dir == NULL || socket_dir[0] == '\0') {
+        socket_dir = "/tmp/.xrdp";
+    }
+
+    socket_name = pa_modargs_get_value(ma, "xrdp_pulse_sink_socket",
+                                       getenv("XRDP_PULSE_SINK_SOCKET"));
+    if (socket_name == NULL || socket_name[0] == '\0')
+    {
+        int display_num = get_display_num_from_display(getenv("DISPLAY"));
+
+        pa_log_debug("Could not obtain sink_socket from environment.");
+        snprintf(default_socket_name, sizeof(default_socket_name),
+                 "xrdp_chansrv_audio_out_socket_%d", display_num);
+        socket_name = default_socket_name;
+    }
+
+    nbytes = strlen(socket_dir) + 1 + strlen(socket_name) + 1;
+    u->sink_socket = pa_xmalloc(nbytes);
+    snprintf(u->sink_socket, nbytes, "%s/%s", socket_dir, socket_name);
+}
+
 int pa__init(pa_module*m) {
     struct userdata *u = NULL;
     pa_sample_spec ss;
@@ -493,6 +578,9 @@ int pa__init(pa_module*m) {
     }
 
     u->sink->parent.process_msg = sink_process_msg;
+#ifdef USE_SET_STATE_IN_IO_THREAD_CB
+    u->sink->set_state_in_io_thread = sink_set_state_in_io_thread_cb;
+#endif
     u->sink->update_requested_latency = sink_update_requested_latency_cb;
     u->sink->userdata = u;
 
@@ -502,9 +590,10 @@ int pa__init(pa_module*m) {
     u->block_usec = BLOCK_USEC;
     pa_log_debug("3 block_usec %llu", (unsigned long long) u->block_usec);
     nbytes = pa_usec_to_bytes(u->block_usec, &u->sink->sample_spec);
+    pa_sink_set_max_rewind(u->sink, nbytes);
     pa_sink_set_max_request(u->sink, nbytes);
 
-    u->display_num = get_display_num_from_display(getenv("DISPLAY"));
+    set_sink_socket(ma, u);
 
     u->fd = -1;
 
@@ -521,6 +610,8 @@ int pa__init(pa_module*m) {
         goto fail;
     }
 
+    pa_sink_set_latency_range(u->sink, 0, BLOCK_USEC);
+
     pa_sink_put(u->sink);
 
     pa_modargs_free(ma);
@@ -575,5 +666,12 @@ void pa__done(pa_module*m) {
         pa_rtpoll_free(u->rtpoll);
     }
 
+    if (u->fd >= 0)
+    {
+        close(u->fd);
+        u->fd = -1;
+    }
+
+    pa_xfree(u->sink_socket);
     pa_xfree(u);
 }
diff --git a/src/module-xrdp-source.c b/src/module-xrdp-source.c
index d31b03b..f0c6aa7 100644
--- a/src/module-xrdp-source.c
+++ b/src/module-xrdp-source.c
@@ -69,7 +69,9 @@ PA_MODULE_USAGE(
         "source_name=<name of source> "
         "channel_map=<channel map> "
         "description=<description for the source> "
-        "latency_time=<latency time in ms>");
+        "latency_time=<latency time in ms> "
+        "xrdp_socket_path=<path to XRDP sockets> "
+        "xrdp_pulse_source_socket=<name of source socket>");
 
 #define DEFAULT_SOURCE_NAME "xrdp-source"
 #define DEFAULT_LATENCY_TIME 10
@@ -92,7 +94,7 @@ struct userdata {
 
     /* xrdp stuff */
     int fd;            /* UDS connection to xrdp chansrv */
-    int display_num;   /* X display number */
+    char *source_socket;
     int want_src_data;
 };
 
@@ -104,10 +106,12 @@ static const char* const valid_modargs[] = {
     "channel_map",
     "description",
     "latency_time",
+    "xrdp_socket_path",
+    "xrdp_pulse_source_socket",
     NULL
 };
 
-static int get_display_num_from_display(char *display_text) ;
+static int get_display_num_from_display(const char *display_text) ;
 
 static int source_process_msg(pa_msgobject *o, int code, void *data,
                               int64_t offset, pa_memchunk *chunk) {
@@ -178,8 +182,6 @@ static int data_get(struct userdata *u, pa_memchunk *chunk) {
     int read_bytes;
     struct sockaddr_un s;
     char *data;
-    char *socket_dir;
-    char *source_socket;
     char buf[11];
     unsigned char ubuf[10];
 
@@ -188,27 +190,7 @@ static int data_get(struct userdata *u, pa_memchunk *chunk) {
         fd = socket(PF_LOCAL, SOCK_STREAM, 0);
         memset(&s, 0, sizeof(s));
         s.sun_family = AF_UNIX;
-        bytes = sizeof(s.sun_path) - 1;
-        socket_dir = getenv("XRDP_SOCKET_PATH");
-        if (socket_dir == NULL || socket_dir[0] == '\0')
-        {
-            socket_dir = "/tmp/.xrdp";
-        }
-        source_socket = getenv("XRDP_PULSE_SOURCE_SOCKET");
-        if (source_socket == NULL || source_socket[0] == '\0')
-        {
-
-            pa_log_debug("Could not obtain source_socket from environment.");
-            /* usually it doesn't reach here. if the socket name is not given
-               via environment variable, use hardcoded name as fallback */
-            snprintf(s.sun_path, bytes,
-                    "%s/xrdp_chansrv_audio_in_socket_%d", socket_dir, u->display_num);
-        }
-        else
-        {
-            pa_log_debug("Obtained source_socket from environment.");
-            snprintf(s.sun_path, bytes, "%s/%s", socket_dir, source_socket);
-        }
+            pa_strlcpy(s.sun_path, u->source_socket, sizeof(s.sun_path));
         pa_log_debug("Trying to connect to %s", s.sun_path);
 
         if (connect(fd, (struct sockaddr *) &s, sizeof(struct sockaddr_un)) != 0) {
@@ -386,6 +368,36 @@ finish:
     pa_log_debug("###### thread shutting down");
 }
 
+static void set_source_socket(pa_modargs *ma, struct userdata *u) {
+    const char *socket_dir;
+    const char *socket_name;
+    char default_socket_name[64];
+    size_t nbytes;
+
+
+    socket_dir = pa_modargs_get_value(ma, "xrdp_socket_path",
+                                      getenv("XRDP_SOCKET_PATH"));
+    if (socket_dir == NULL || socket_dir[0] == '\0') {
+        socket_dir = "/tmp/.xrdp";
+    }
+
+    socket_name = pa_modargs_get_value(ma, "xrdp_pulse_source_socket",
+                                       getenv("XRDP_PULSE_SOURCE_SOCKET"));
+    if (socket_name == NULL || socket_name[0] == '\0')
+    {
+        int display_num = get_display_num_from_display(getenv("DISPLAY"));
+
+        pa_log_debug("Could not obtain source_socket from environment.");
+        snprintf(default_socket_name, sizeof(default_socket_name),
+                 "xrdp_chansrv_audio_out_socket_%d", display_num);
+        socket_name = default_socket_name;
+    }
+
+    nbytes = strlen(socket_dir) + 1 + strlen(socket_name) + 1;
+    u->source_socket = pa_xmalloc(nbytes);
+    snprintf(u->source_socket, nbytes, "%s/%s", socket_dir, socket_name);
+}
+
 int pa__init(pa_module *m) {
     struct userdata *u = NULL;
     pa_sample_spec ss;
@@ -473,6 +485,8 @@ int pa__init(pa_module *m) {
     u->source->thread_info.max_rewind =
         pa_usec_to_bytes(u->block_usec, &u->source->sample_spec);
 
+    set_source_socket(ma, u);
+
     u->fd = -1;
 
 #if defined(PA_CHECK_VERSION)
@@ -492,8 +506,6 @@ int pa__init(pa_module *m) {
 
     pa_modargs_free(ma);
 
-    u->display_num = get_display_num_from_display(getenv("DISPLAY"));
-
     return 0;
 
 fail:
@@ -529,10 +541,17 @@ void pa__done(pa_module*m) {
     if (u->rtpoll)
         pa_rtpoll_free(u->rtpoll);
 
+    if (u->fd >= 0)
+    {
+        close(u->fd);
+        u->fd = -1;
+    }
+
+    pa_xfree(u->source_socket);
     pa_xfree(u);
 }
 
-static int get_display_num_from_display(char *display_text) {
+static int get_display_num_from_display(const char *display_text) {
     int index;
     int mode;
     int host_index;