diff --git a/src/context.c b/src/context.c
index 123459a26..69788cf28 100644
--- a/src/context.c
+++ b/src/context.c
@@ -139,7 +139,7 @@ struct global *pa_context_find_global_by_name(pa_context *c, uint32_t mask, cons
 		    (str = pw_properties_get(g->props, PW_KEY_NODE_NAME)) != NULL &&
 		    strcmp(str, name) == 0)
 			return g;
-		if (g->id == id)
+		if (g->id == id || (g->id == (id & PA_IDX_MASK_DSP)))
 			return g;
 	}
 	return NULL;
@@ -163,14 +163,8 @@ struct global *pa_context_find_linked(pa_context *c, uint32_t idx)
 			f = pa_context_find_global(c, g->link_info.src->parent_id);
 		else
 			continue;
-
 		if (f == NULL)
 			continue;
-		if (f->mask & PA_SUBSCRIPTION_MASK_DSP) {
-			if (!(f->mask & PA_SUBSCRIPTION_MASK_SOURCE) ||
-			    g->link_info.dst->parent_id != idx)
-				f = pa_context_find_global(c, f->dsp_info.session);
-		}
 		return f;
 	}
 	return NULL;
@@ -178,12 +172,19 @@ struct global *pa_context_find_linked(pa_context *c, uint32_t idx)
 
 static void emit_event(pa_context *c, struct global *g, pa_subscription_event_type_t event)
 {
-	if (c->subscribe_mask & g->mask) {
-		if (c->subscribe_callback) {
-			pw_log_debug("context %p: obj %d: emit %d:%d", c, g->id, event, g->event);
+	if (c->subscribe_callback && (c->subscribe_mask & g->mask)) {
+		pw_log_debug("context %p: obj %d: emit %d:%d", c, g->id, event, g->event);
+		c->subscribe_callback(c,
+				event | g->event,
+				g->id,
+				c->subscribe_userdata);
+
+		if (g->mask == (PA_SUBSCRIPTION_MASK_SINK | PA_SUBSCRIPTION_MASK_SOURCE)) {
+			pw_log_debug("context %p: obj %d: emit %d:%d", c, g->node_info.monitor,
+					event, PA_SUBSCRIPTION_EVENT_SOURCE);
 			c->subscribe_callback(c,
-					event | g->event,
-					g->id,
+					event | PA_SUBSCRIPTION_EVENT_SOURCE,
+					g->node_info.monitor,
 					c->subscribe_userdata);
 		}
 	}
@@ -518,31 +519,28 @@ static int set_mask(pa_context *c, struct global *g)
 			return 0;
 
 		if (strcmp(str, "Audio/Sink") == 0) {
+			return 0;
 			pw_log_debug("found sink %d", g->id);
 			g->mask = PA_SUBSCRIPTION_MASK_SINK;
 			g->event = PA_SUBSCRIPTION_EVENT_SINK;
 			g->node_info.monitor = SPA_ID_INVALID;
 		}
 		else if (strcmp(str, "Audio/DSP/Playback") == 0) {
-			if ((str = pw_properties_get(g->props, PW_KEY_NODE_SESSION)) == NULL)
-				return 0;
-			pw_log_debug("found monitor %d", g->id);
-			g->mask = PA_SUBSCRIPTION_MASK_DSP_SINK | PA_SUBSCRIPTION_MASK_SOURCE;
-			g->event = PA_SUBSCRIPTION_EVENT_SOURCE;
-			g->dsp_info.session = pw_properties_parse_int(str);
-			if ((f = pa_context_find_global(c, g->dsp_info.session)) != NULL)
-				f->node_info.monitor = g->id;
+			pw_log_debug("found playback/monitor DSP %d", g->id);
+			g->mask = PA_SUBSCRIPTION_MASK_SINK | PA_SUBSCRIPTION_MASK_SOURCE;
+			g->event = PA_SUBSCRIPTION_EVENT_SINK;
+			g->node_info.monitor = g->id | PA_IDX_FLAG_DSP;
 		}
 		else if (strcmp(str, "Audio/Source") == 0) {
+			return 0;
 			pw_log_debug("found source %d", g->id);
 			g->mask = PA_SUBSCRIPTION_MASK_SOURCE;
 			g->event = PA_SUBSCRIPTION_EVENT_SOURCE;
 		}
 		else if (strcmp(str, "Audio/DSP/Capture") == 0) {
-			if ((str = pw_properties_get(g->props, PW_KEY_NODE_SESSION)) == NULL)
-				return 0;
-			g->mask = PA_SUBSCRIPTION_MASK_DSP_SOURCE;
-			g->dsp_info.session = pw_properties_parse_int(str);
+			pw_log_debug("found capture DSP %d", g->id);
+			g->mask = PA_SUBSCRIPTION_MASK_SOURCE;
+			g->event = PA_SUBSCRIPTION_EVENT_SOURCE;
 		}
 		else if (strcmp(str, "Stream/Output/Audio") == 0) {
 			pw_log_debug("found sink input %d", g->id);
diff --git a/src/internal.h b/src/internal.h
index 12d458673..a6a8048ea 100644
--- a/src/internal.h
+++ b/src/internal.h
@@ -210,9 +210,8 @@ struct param {
 	void *param;
 };
 
-#define PA_SUBSCRIPTION_MASK_DSP_SINK	0x1000U
-#define PA_SUBSCRIPTION_MASK_DSP_SOURCE	0x2000U
-#define PA_SUBSCRIPTION_MASK_DSP	(PA_SUBSCRIPTION_MASK_DSP_SINK | PA_SUBSCRIPTION_MASK_DSP_SOURCE)
+#define PA_IDX_FLAG_DSP		0x800000U
+#define PA_IDX_MASK_DSP		0x7fffffU
 
 struct global {
 	struct spa_list link;
@@ -242,10 +241,6 @@ struct global {
 			struct global *src;
 			struct global *dst;
 		} link_info;
-		/* for dsp source and sink */
-		struct {
-			uint32_t session;
-		} dsp_info;
 		/* for sink/source */
 		struct {
 			uint32_t monitor;
diff --git a/src/introspect.c b/src/introspect.c
index 8f22d2785..ba3105756 100644
--- a/src/introspect.c
+++ b/src/introspect.c
@@ -507,9 +507,10 @@ static void source_callback(struct source_data *d)
 	i.owner_module = g->parent_id;
 	pa_cvolume_set(&i.volume, 2, g->node_info.volume * PA_VOLUME_NORM);
 	i.mute = g->node_info.mute;
-	if (g->mask & PA_SUBSCRIPTION_MASK_DSP_SINK) {
-		i.monitor_of_sink = g->dsp_info.session;
+	if (g->mask & PA_SUBSCRIPTION_MASK_SINK) {
+		i.monitor_of_sink = g->id;
 		i.monitor_of_sink_name = "unknown";
+		i.index = g->node_info.monitor;
 	} else {
 		i.monitor_of_sink = PA_INVALID_INDEX;
 		i.monitor_of_sink_name = NULL;
@@ -588,9 +589,12 @@ pa_operation* pa_context_get_source_info_by_index(pa_context *c, uint32_t idx, p
 
 	PA_CHECK_VALIDITY_RETURN_NULL(c, idx != PA_INVALID_INDEX, PA_ERR_INVALID);
 
-	if ((g = pa_context_find_global(c, idx)) == NULL)
-		return NULL;
-	if (!(g->mask & PA_SUBSCRIPTION_MASK_SOURCE))
+	pw_log_debug("context %p: index %d", c, idx);
+
+	if (((g = pa_context_find_global(c, idx)) == NULL ||
+	    !(g->mask & PA_SUBSCRIPTION_MASK_SOURCE)) &&
+	    (((g = pa_context_find_global(c, idx & PA_IDX_MASK_DSP)) == NULL ||
+	    !(g->mask & PA_SUBSCRIPTION_MASK_SOURCE))))
 		return NULL;
 
 	o = pa_operation_new(c, NULL, source_info, sizeof(struct source_data));
diff --git a/src/mainloop.c b/src/mainloop.c
index dbc4d7851..533e4deb5 100644
--- a/src/mainloop.c
+++ b/src/mainloop.c
@@ -286,10 +286,18 @@ int pa_mainloop_prepare(pa_mainloop *m, int timeout)
 SPA_EXPORT
 int pa_mainloop_poll(pa_mainloop *m)
 {
+	int res;
+
 	if (m->quit)
 		return -2;
 
-	return m->n_events = pw_loop_iterate(m->loop, m->timeout);
+	res = pw_loop_iterate(m->loop, m->timeout);
+	if (res < 0) {
+		if (res == -EINTR)
+			res = 0;
+	}
+	m->n_events = res;
+	return res;
 }
 
 SPA_EXPORT
diff --git a/src/stream.c b/src/stream.c
index 5673bd7bf..d81022ea8 100644
--- a/src/stream.c
+++ b/src/stream.c
@@ -197,7 +197,16 @@ static void configure_device(pa_stream *s)
 		s->device_name = NULL;
 	}
 	else {
-		s->device_index = g->id;
+		if (s->direction == PA_STREAM_RECORD) {
+			if (g->mask == (PA_SUBSCRIPTION_MASK_SINK | PA_SUBSCRIPTION_MASK_SOURCE))
+				s->device_index = g->node_info.monitor;
+			else
+				s->device_index = g->id;
+		}
+		else {
+			s->device_index = g->id;
+		}
+
 		if ((str = pw_properties_get(g->props, PW_KEY_NODE_NAME)) == NULL)
 			s->device_name = strdup("unknown");
 		else