Currently the pipe-source does not produce any data if no
writer is connected. This patch enable silence generator
when last writer closed pipe. It will stop automatically
when any data appears.
---
 src/modules/module-pipe-source.c | 113 ++++++++++++++++++++++++++++++---------
 1 file changed, 87 insertions(+), 26 deletions(-)

diff --git a/src/modules/module-pipe-source.c b/src/modules/module-pipe-source.c
index f8284c1..b7eb4bf 100644
--- a/src/modules/module-pipe-source.c
+++ b/src/modules/module-pipe-source.c
@@ -33,6 +33,7 @@
 #include <sys/filio.h>
 #endif
 
+#include <pulse/rtclock.h>
 #include <pulse/xmalloc.h>
 
 #include <pulsecore/core-error.h>
@@ -71,7 +72,11 @@ struct userdata {
     pa_thread_mq thread_mq;
     pa_rtpoll *rtpoll;
 
+    pa_usec_t timestamp;
+    pa_usec_t latency;
+
     char *filename;
+    int corkfd;
     int fd;
 
     pa_memchunk memchunk;
@@ -90,6 +95,7 @@ static const char* const valid_modargs[] = {
     NULL
 };
 
+/* Called from thread context */
 static int source_process_msg(
         pa_msgobject *o,
         int code,
@@ -101,17 +107,28 @@ static int source_process_msg(
 
     switch (code) {
 
+        case PA_SOURCE_MESSAGE_SET_STATE:
+
+            if (u->corkfd >= 0 && PA_SOURCE_IS_OPENED(PA_PTR_TO_UINT(data)))
+                u->timestamp = pa_rtclock_now();
+
+            break;
+
         case PA_SOURCE_MESSAGE_GET_LATENCY: {
-            size_t n = 0;
+            int64_t latency = 0;
 
+            if (u->corkfd >= 0)
+                latency = (int64_t)pa_rtclock_now() - (int64_t)u->timestamp;
 #ifdef FIONREAD
-            int l;
+            else {
+                int l;
 
-            if (ioctl(u->fd, FIONREAD, &l) >= 0 && l > 0)
-                n = (size_t) l;
+                if (ioctl(u->fd, FIONREAD, &l) >= 0 && l > 0)
+                    latency = pa_bytes_to_usec((size_t) l, 
&u->source->sample_spec);
+            }
 #endif
 
-            *((int64_t*) data) = pa_bytes_to_usec(n, &u->source->sample_spec);
+            *((int64_t*) data) = latency;
             return 0;
         }
     }
@@ -129,6 +146,10 @@ static void thread_func(void *userdata) {
 
     pa_thread_mq_install(&u->thread_mq);
 
+    /* Close our writer here to start silence generation or suspend source if 
no writers left */
+    pa_assert_se(pa_close(u->corkfd) == 0);
+    u->corkfd = -1;
+
     for (;;) {
         int ret;
         struct pollfd *pollfd;
@@ -136,7 +157,7 @@ static void thread_func(void *userdata) {
         pollfd = pa_rtpoll_item_get_pollfd(u->rtpoll_item, NULL);
 
         /* Try to read some data and pass it on to the source driver */
-        if (u->source->thread_info.state == PA_SOURCE_RUNNING && 
pollfd->revents) {
+        if (pollfd->revents) {
             ssize_t l;
             void *p;
 
@@ -151,47 +172,78 @@ static void thread_func(void *userdata) {
             l = pa_read(u->fd, (uint8_t*) p + u->memchunk.index, 
pa_memblock_get_length(u->memchunk.memblock) - u->memchunk.index, &read_type);
             pa_memblock_release(u->memchunk.memblock);
 
-            pa_assert(l != 0); /* EOF cannot happen, since we opened the fifo 
for both reading and writing */
+            if (PA_LIKELY(l > 0)) {
+                if (PA_SOURCE_IS_OPENED(u->source->thread_info.state)) {
+                    u->memchunk.length = (size_t) l;
+                    pa_source_post(u->source, &u->memchunk);
+                    u->memchunk.index += (size_t) l;
+
+                    if (u->memchunk.index >= 
pa_memblock_get_length(u->memchunk.memblock)) {
+                        pa_memblock_unref(u->memchunk.memblock);
+                        pa_memchunk_reset(&u->memchunk);
+                    }
+                }
 
-            if (l < 0) {
+                if (u->corkfd >= 0) {
+                    pa_assert_se(pa_close(u->corkfd) == 0);
+                    u->corkfd = -1;
 
+                    pa_rtpoll_set_timer_disabled(u->rtpoll);
+                }
+            } else if (l == 0) {
+                if (u->corkfd < 0) {
+                    pa_log_debug("There are no writers left");
+
+                    if ((u->corkfd = pa_open_cloexec(u->filename, O_WRONLY, 
0)) < 0) {
+                        pa_log("open('%s'): %s", u->filename, 
pa_cstrerror(errno));
+                        goto fail;
+                    }
+
+                    u->latency = pa_bytes_to_usec(pa_pipe_buf(u->fd), 
&u->source->sample_spec);
+                    u->timestamp = pa_rtclock_now();
+                }
+            } else {
                 if (errno == EINTR)
                     continue;
                 else if (errno != EAGAIN) {
                     pa_log("Failed to read data from FIFO: %s", 
pa_cstrerror(errno));
                     goto fail;
                 }
+            }
+        }
 
-            } else {
+        if (u->corkfd >= 0 && 
PA_SOURCE_IS_OPENED(u->source->thread_info.state)) {
+            pa_usec_t now;
+            size_t l;
 
-                u->memchunk.length = (size_t) l;
-                pa_source_post(u->source, &u->memchunk);
-                u->memchunk.index += (size_t) l;
+            now = pa_rtclock_now();
+            l = pa_usec_to_bytes(now - u->timestamp, &u->source->sample_spec);
 
-                if (u->memchunk.index >= 
pa_memblock_get_length(u->memchunk.memblock)) {
-                    pa_memblock_unref(u->memchunk.memblock);
-                    pa_memchunk_reset(&u->memchunk);
-                }
+            if (l > 0) {
+                pa_memchunk chunk = {
+                    .index = 0,
+                    .length = l,
+                    .memblock = pa_memblock_new(u->core->mempool, l)
+                };
 
-                pollfd->revents = 0;
+                pa_silence_memchunk(&chunk, &u->source->sample_spec);
+                pa_source_post(u->source, &chunk);
+                pa_memblock_unref(chunk.memblock);
+
+                u->timestamp += pa_bytes_to_usec(l, &u->source->sample_spec);
             }
+
+            pa_rtpoll_set_timer_absolute(u->rtpoll, u->timestamp + u->latency);
         }
 
         /* Hmm, nothing to do. Let's sleep */
-        pollfd->events = (short) (u->source->thread_info.state == 
PA_SOURCE_RUNNING ? POLLIN : 0);
+        pollfd->events = POLLIN;
 
         if ((ret = pa_rtpoll_run(u->rtpoll)) < 0)
             goto fail;
 
         if (ret == 0)
             goto finish;
-
-        pollfd = pa_rtpoll_item_get_pollfd(u->rtpoll_item, NULL);
-
-        if (pollfd->revents & ~POLLIN) {
-            pa_log("FIFO shutdown.");
-            goto fail;
-        }
     }
 
 fail:
@@ -244,7 +296,13 @@ int pa__init(pa_module *m) {
         pa_log("mkfifo('%s'): %s", u->filename, pa_cstrerror(errno));
         goto fail;
     }
-    if ((u->fd = pa_open_cloexec(u->filename, O_RDWR, 0)) < 0) {
+
+    if ((u->corkfd = pa_open_cloexec(u->filename, O_RDWR, 0)) < 0) {
+        pa_log("open('%s'): %s", u->filename, pa_cstrerror(errno));
+        goto fail;
+    }
+
+    if ((u->fd = pa_open_cloexec(u->filename, O_RDONLY, 0)) < 0) {
         pa_log("open('%s'): %s", u->filename, pa_cstrerror(errno));
         goto fail;
     }
@@ -360,6 +418,9 @@ void pa__done(pa_module *m) {
         pa_xfree(u->filename);
     }
 
+    if (u->corkfd >= 0)
+        pa_assert_se(pa_close(u->corkfd) == 0);
+
     if (u->fd >= 0)
         pa_assert_se(pa_close(u->fd) == 0);
 
-- 
1.8.3.1

_______________________________________________
pulseaudio-discuss mailing list
pulseaudio-discuss@lists.freedesktop.org
https://lists.freedesktop.org/mailman/listinfo/pulseaudio-discuss

Reply via email to