[PATCH] pipe-sink: auto-drain the pipe on sink resume

[samo_pogacnik@xxxxxxx (Samo Pogačnik)]

Added option auto_drain_pipe_on_resume to enable draining any remaining
data from the pipe upon every pipe-sink resume out of suspend.
---
 src/modules/module-pipe-sink.c | 37 ++++++++++++++++++++++++++++++++++++-
 1 file changed, 36 insertions(+), 1 deletion(-)

diff --git a/src/modules/module-pipe-sink.c b/src/modules/module-pipe-sink.c
index 995785e..10ce882 100644
--- a/src/modules/module-pipe-sink.c
+++ b/src/modules/module-pipe-sink.c
@@ -62,6 +62,7 @@ PA_MODULE_USAGE(
         "channels=<number of channels> "
         "channel_map=<channel map> "
         "use_system_clock_for_timing=<yes or no> "
+        "auto_drain_pipe_on_resume=<yes or no> "
 );
 
 #define DEFAULT_FILE_NAME "fifo_output"
@@ -82,16 +83,19 @@ struct userdata {
     size_t buffer_size;
     size_t bytes_dropped;
     bool fifo_error;
+    uint8_t *drain_buffer;
 
     pa_memchunk memchunk;
 
     pa_rtpoll_item *rtpoll_item;
 
+    int read_type;
     int write_type;
     pa_usec_t block_usec;
     pa_usec_t timestamp;
 
     bool use_system_clock_for_timing;
+    bool auto_drain_pipe_on_resume;
 };
 
 static const char* const valid_modargs[] = {
@@ -103,17 +107,39 @@ static const char* const valid_modargs[] = {
     "channels",
     "channel_map",
     "use_system_clock_for_timing",
+    "auto_drain_pipe_on_resume",
     NULL
 };
 
+static ssize_t drain_pipe_sink(struct userdata *u) {
+    ssize_t l, drained = 0;
+
+    pa_assert(u);
+
+    do {
+        l = pa_read(u->fd, u->drain_buffer, u->buffer_size, &u->read_type);
+
+        if (l > 0)
+            drained += l;
+
+    } while ((l > 0) || (errno == EINTR));
+
+    return drained;
+}
+
 static int sink_process_msg(pa_msgobject *o, int code, void *data, int64_t offset, pa_memchunk *chunk) {
     struct userdata *u = PA_SINK(o)->userdata;
 
     switch (code) {
         case PA_SINK_MESSAGE_SET_STATE:
             if (u->sink->thread_info.state == PA_SINK_SUSPENDED || u->sink->thread_info.state == PA_SINK_INIT) {
-                if (PA_SINK_IS_OPENED(PA_PTR_TO_UINT(data)))
+                if (PA_SINK_IS_OPENED(PA_PTR_TO_UINT(data))) {
                     u->timestamp = pa_rtclock_now();
+                    if (u->auto_drain_pipe_on_resume) {
+                        ssize_t d = drain_pipe_sink(u);
+                        pa_log_debug("Pipe-sink resume from suspend: auto-drained %zd bytes from the pipe", d);
+                    }
+                }
             } else if (u->sink->thread_info.state == PA_SINK_RUNNING || u->sink->thread_info.state == PA_SINK_IDLE) {
                 if (PA_PTR_TO_UINT(data) == PA_SINK_SUSPENDED) {
                     /* Clear potential FIFO error flag */
@@ -443,11 +469,17 @@ int pa__init(pa_module *m) {
         goto fail;
     }
 
+    if (pa_modargs_get_value_boolean(ma, "auto_drain_pipe_on_resume", &u->auto_drain_pipe_on_resume) < 0) {
+        pa_log("Failed to parse auto_drain_pipe_on_resume argument.");
+        goto fail;
+    }
+
     if (pa_thread_mq_init(&u->thread_mq, m->core->mainloop, u->rtpoll) < 0) {
         pa_log("pa_thread_mq_init() failed.");
         goto fail;
     }
 
+    u->read_type = 0;
     u->write_type = 0;
 
     u->filename = pa_runtime_path(pa_modargs_get_value(ma, "file", DEFAULT_FILE_NAME));
@@ -525,6 +557,9 @@ int pa__init(pa_module *m) {
     }
     pa_sink_set_max_request(u->sink, u->buffer_size);
 
+    if (u->auto_drain_pipe_on_resume)
+        u->drain_buffer = (uint8_t *)pa_xmalloc(u->buffer_size);
+
     u->rtpoll_item = pa_rtpoll_item_new(u->rtpoll, PA_RTPOLL_NEVER, 1);
     pollfd = pa_rtpoll_item_get_pollfd(u->rtpoll_item, NULL);
     pollfd->fd = u->fd;
-- 
2.7.4

regards, Samo