5 files changed, 131 insertions, 5 deletions

diff --git a/src/pulsecore/sink-input.h b/src/pulsecore/sink-input.h
index c4e65b50..a101828f 100644
--- a/src/pulsecore/sink-input.h
+++ b/src/pulsecore/sink-input.h
@@ -91,7 +91,10 @@ struct pa_sink_input {
     void (*drop) (pa_sink_input *i, size_t length);
 
     /* If non-NULL this function is called when the input is first
-     * connected to a sink. Called from IO thread context */
+     * connected to a sink or when the rtpoll/asyncmsgq fields
+     * change. You usually don't need to implement this function
+     * unless you rewrite a sink that is piggy-backed onto
+     * another. Called from IO thread context */
     void (*attach) (pa_sink_input *i);           /* may be NULL */ 
 
     /* If non-NULL this function is called when the output is
diff --git a/src/pulsecore/sink.c b/src/pulsecore/sink.c
index a7ed5a40..9b191886 100644
--- a/src/pulsecore/sink.c
+++ b/src/pulsecore/sink.c
@@ -901,6 +901,20 @@ int pa_sink_process_msg(pa_msgobject *o, int code, void *userdata, int64_t offse
             s->thread_info.state = PA_PTR_TO_UINT(userdata);
             return 0;
 
+        case PA_SINK_MESSAGE_DETACH:
+
+            /* We're detaching all our input streams so that the
+             * asyncmsgq and rtpoll fields can be changed without
+             * problems */
+            pa_sink_detach_within_thread(s);
+            break;
+
+        case PA_SINK_MESSAGE_ATTACH:
+
+            /* Reattach all streams */
+            pa_sink_attach_within_thread(s);
+            break;
+            
         case PA_SINK_MESSAGE_GET_LATENCY:
         case PA_SINK_MESSAGE_MAX:
             ;
@@ -922,3 +936,47 @@ int pa_sink_suspend_all(pa_core *c, int suspend) {
     return ret;
 }
 
+void pa_sink_detach(pa_sink *s) {
+    pa_sink_assert_ref(s);
+    pa_assert(PA_SINK_LINKED(s->state));
+
+    pa_asyncmsgq_send(s->asyncmsgq, PA_MSGOBJECT(s), PA_SINK_MESSAGE_DETACH, NULL, 0, NULL);
+}
+
+void pa_sink_attach(pa_sink *s) {
+    pa_sink_assert_ref(s);
+    pa_assert(PA_SINK_LINKED(s->state));
+
+    pa_asyncmsgq_send(s->asyncmsgq, PA_MSGOBJECT(s), PA_SINK_MESSAGE_ATTACH, NULL, 0, NULL);
+}
+
+void pa_sink_detach_within_thread(pa_sink *s) {
+    pa_sink_input *i;
+    void *state = NULL;
+
+    pa_sink_assert_ref(s);
+    pa_assert(PA_SINK_LINKED(s->thread_info.state));
+
+    while ((i = pa_hashmap_iterate(s->thread_info.inputs, &state, NULL)))
+        if (i->detach)
+            i->detach(i);
+
+    if (s->monitor_source)
+        pa_source_detach_within_thread(s->monitor_source);
+}
+
+void pa_sink_attach_within_thread(pa_sink *s) {
+    pa_sink_input *i;
+    void *state = NULL;
+
+    pa_sink_assert_ref(s);
+    pa_assert(PA_SINK_LINKED(s->thread_info.state));
+
+    while ((i = pa_hashmap_iterate(s->thread_info.inputs, &state, NULL)))
+        if (i->attach)
+            i->attach(i);
+
+    if (s->monitor_source)
+        pa_source_attach_within_thread(s->monitor_source);
+}
+
diff --git a/src/pulsecore/sink.h b/src/pulsecore/sink.h
index b3fcff49..be883ed6 100644
--- a/src/pulsecore/sink.h
+++ b/src/pulsecore/sink.h
@@ -122,6 +122,8 @@ typedef enum pa_sink_message {
     PA_SINK_MESSAGE_SET_STATE,
     PA_SINK_MESSAGE_PING,
     PA_SINK_MESSAGE_REMOVE_INPUT_AND_BUFFER,
+    PA_SINK_MESSAGE_ATTACH,
+    PA_SINK_MESSAGE_DETACH,
     PA_SINK_MESSAGE_MAX
 } pa_sink_message_t;
 
@@ -143,6 +145,9 @@ void pa_sink_set_description(pa_sink *s, const char *description);
 void pa_sink_set_asyncmsgq(pa_sink *s, pa_asyncmsgq *q);
 void pa_sink_set_rtpoll(pa_sink *s, pa_rtpoll *p);
 
+void pa_sink_detach(pa_sink *s);
+void pa_sink_attach(pa_sink *s);
+
 /* May be called by everyone, from main context */
 
 pa_usec_t pa_sink_get_latency(pa_sink *s);
@@ -173,8 +178,9 @@ void pa_sink_render_into_full(pa_sink *s, pa_memchunk *target);
 
 void pa_sink_skip(pa_sink *s, size_t length);
 
-int pa_sink_process_inputs(pa_sink *s);
-
 int pa_sink_process_msg(pa_msgobject *o, int code, void *userdata, int64_t offset, pa_memchunk *chunk);
 
+void pa_sink_attach_within_thread(pa_sink *s);
+void pa_sink_detach_within_thread(pa_sink *s);
+
 #endif
diff --git a/src/pulsecore/source.c b/src/pulsecore/source.c
index 34e023de..63ff9d7d 100644
--- a/src/pulsecore/source.c
+++ b/src/pulsecore/source.c
@@ -483,6 +483,20 @@ int pa_source_process_msg(pa_msgobject *object, int code, void *userdata, int64_
         case PA_SOURCE_MESSAGE_SET_STATE:
             s->thread_info.state = PA_PTR_TO_UINT(userdata);
             return 0;
+
+        case PA_SOURCE_MESSAGE_DETACH:
+
+            /* We're detaching all our output streams so that the
+             * asyncmsgq and rtpoll fields can be changed without
+             * problems */
+            pa_source_detach_within_thread(s);
+            break;
+
+        case PA_SOURCE_MESSAGE_ATTACH:
+
+            /* Reattach all streams */
+            pa_source_attach_within_thread(s);
+            break;
             
         case PA_SOURCE_MESSAGE_GET_LATENCY:
         case PA_SOURCE_MESSAGE_MAX:
@@ -504,3 +518,42 @@ int pa_source_suspend_all(pa_core *c, int suspend) {
 
     return ret;
 }
+
+void pa_source_detach(pa_source *s) {
+    pa_source_assert_ref(s);
+    pa_assert(PA_SOURCE_LINKED(s->state));
+
+    pa_asyncmsgq_send(s->asyncmsgq, PA_MSGOBJECT(s), PA_SOURCE_MESSAGE_DETACH, NULL, 0, NULL);
+}
+
+void pa_source_attach(pa_source *s) {
+    pa_source_assert_ref(s);
+    pa_assert(PA_SOURCE_LINKED(s->state));
+
+    pa_asyncmsgq_send(s->asyncmsgq, PA_MSGOBJECT(s), PA_SOURCE_MESSAGE_ATTACH, NULL, 0, NULL);
+}
+
+void pa_source_detach_within_thread(pa_source *s) {
+    pa_source_output *o;
+    void *state = NULL;
+
+    pa_source_assert_ref(s);
+    pa_assert(PA_SOURCE_LINKED(s->thread_info.state));
+
+    while ((o = pa_hashmap_iterate(s->thread_info.outputs, &state, NULL)))
+        if (o->detach)
+            o->detach(o);
+}
+
+void pa_source_attach_within_thread(pa_source *s) {
+    pa_source_output *o;
+    void *state = NULL;
+
+    pa_source_assert_ref(s);
+    pa_assert(PA_SOURCE_LINKED(s->thread_info.state));
+
+    while ((o = pa_hashmap_iterate(s->thread_info.outputs, &state, NULL)))
+        if (o->attach)
+            o->attach(o);
+
+}
diff --git a/src/pulsecore/source.h b/src/pulsecore/source.h
index ddc66156..0fd14860 100644
--- a/src/pulsecore/source.h
+++ b/src/pulsecore/source.h
@@ -121,6 +121,8 @@ typedef enum pa_source_message {
     PA_SOURCE_MESSAGE_GET_LATENCY,
     PA_SOURCE_MESSAGE_SET_STATE,
     PA_SOURCE_MESSAGE_PING,
+    PA_SOURCE_MESSAGE_ATTACH,
+    PA_SOURCE_MESSAGE_DETACH,
     PA_SOURCE_MESSAGE_MAX
 } pa_source_message_t;
 
@@ -142,6 +144,9 @@ void pa_source_set_description(pa_source *s, const char *description);
 void pa_source_set_asyncmsgq(pa_source *s, pa_asyncmsgq *q);
 void pa_source_set_rtpoll(pa_source *s, pa_rtpoll *p);
 
+void pa_source_detach(pa_source *s);
+void pa_source_attach(pa_source *s);
+
 /* May be called by everyone, from main context */
 
 pa_usec_t pa_source_get_latency(pa_source *s);
@@ -164,8 +169,9 @@ unsigned pa_source_used_by(pa_source *s);
 
 void pa_source_post(pa_source*s, const pa_memchunk *b);
 
-int pa_source_process_outputs(pa_source *o);
-
 int pa_source_process_msg(pa_msgobject *o, int code, void *userdata, int64_t, pa_memchunk *chunk);
 
+void pa_source_attach_within_thread(pa_source *s);
+void pa_source_detach_within_thread(pa_source *s);
+
 #endif