diff --git a/src/context.c b/src/context.c
index 6a317907e..545df2414 100644
--- a/src/context.c
+++ b/src/context.c
@@ -104,7 +104,7 @@ struct global *pa_context_find_global_by_name(pa_context *c, uint32_t mask, cons
 	uint32_t id = atoi(name);
 
 	spa_list_for_each(g, &c->globals, link) {
-		if (!(g->mask & mask))
+		if ((g->mask & mask) == 0)
 			continue;
 		if (g->props != NULL &&
 		    (str = pw_properties_get(g->props, "node.name")) != NULL &&
@@ -138,7 +138,9 @@ struct global *pa_context_find_linked(pa_context *c, uint32_t idx)
 		if (f == NULL)
 			continue;
 		if (f->mask & PA_SUBSCRIPTION_MASK_DSP) {
-			f = pa_context_find_global(c, f->dsp_info.session);
+			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;
 	}
@@ -148,6 +150,7 @@ struct global *pa_context_find_linked(pa_context *c, uint32_t idx)
 static int set_mask(pa_context *c, struct global *g)
 {
 	const char *str;
+	struct global *f;
 
 	switch (g->type) {
 	case PW_TYPE_INTERFACE_Device:
@@ -172,12 +175,16 @@ static int set_mask(pa_context *c, struct global *g)
 		if (strcmp(str, "Audio/Sink") == 0) {
 			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, "node.session")) == NULL)
 				return 0;
-			g->mask = PA_SUBSCRIPTION_MASK_DSP_SINK;
+			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;
 		}
 		else if (strcmp(str, "Audio/Source") == 0) {
 			g->mask = PA_SUBSCRIPTION_MASK_SOURCE;
diff --git a/src/internal.h b/src/internal.h
index 712868e03..ab99ec20e 100644
--- a/src/internal.h
+++ b/src/internal.h
@@ -222,6 +222,10 @@ struct global {
 		struct {
 			uint32_t session;
 		} dsp_info;
+		/* for sink/source */
+		struct {
+			uint32_t monitor;
+		} node_info;
 		/* for devices */
 		struct {
 			struct pw_array profiles;
diff --git a/src/introspect.c b/src/introspect.c
index 358e4a938..ec3809436 100644
--- a/src/introspect.c
+++ b/src/introspect.c
@@ -334,6 +334,8 @@ static void sink_callback(struct sink_data *d)
 	pa_format_info ii[1];
 	pa_format_info *ip[1];
 
+	pw_log_debug("sink %d %s monitor %d", g->id, info->name, g->node_info.monitor);
+
 	spa_zero(i);
 	i.name = info->name;
 	i.index = g->id;
@@ -345,7 +347,7 @@ static void sink_callback(struct sink_data *d)
 	i.owner_module = g->parent_id;
 	pa_cvolume_set(&i.volume, 2, PA_VOLUME_NORM);
 	i.mute = false;
-	i.monitor_source = PA_INVALID_INDEX;
+	i.monitor_source = g->node_info.monitor;
 	i.monitor_source_name = "unknown";
 	i.latency = 0;
 	i.driver = "PipeWire";
@@ -565,8 +567,13 @@ static void source_callback(struct source_data *d)
 	i.owner_module = g->parent_id;
 	pa_cvolume_set(&i.volume, 2, PA_VOLUME_NORM);
 	i.mute = false;
-	i.monitor_of_sink = PA_INVALID_INDEX;
-	i.monitor_of_sink_name = "unknown";
+	if (g->mask & PA_SUBSCRIPTION_MASK_DSP_SINK) {
+		i.monitor_of_sink = g->dsp_info.session;
+		i.monitor_of_sink_name = "unknown";
+	} else {
+		i.monitor_of_sink = PA_INVALID_INDEX;
+		i.monitor_of_sink_name = NULL;
+	}
 	i.latency = 0;
 	i.driver = "PipeWire";
 	i.flags = 0;
@@ -1473,8 +1480,8 @@ pa_operation* pa_context_set_sink_input_volume(pa_context *c, uint32_t idx, cons
 		pw_node_proxy_set_param((struct pw_node_proxy*)g->proxy,
 			SPA_PARAM_Props, 0,
 			spa_pod_builder_object(&b,
-				SPA_TYPE_OBJECT_Props, SPA_PARAM_Props,
-				SPA_PROP_volume,        &SPA_POD_Float(v),
+				SPA_TYPE_OBJECT_Props,	SPA_PARAM_Props,
+				SPA_PROP_volume,	&SPA_POD_Float(v),
 				0));
 	}
 	o = pa_operation_new(c, NULL, on_success, sizeof(struct success_ack));
@@ -1511,7 +1518,7 @@ pa_operation* pa_context_set_sink_input_mute(pa_context *c, uint32_t idx, int mu
 		pw_node_proxy_set_param((struct pw_node_proxy*)g->proxy,
 			SPA_PARAM_Props, 0,
 			spa_pod_builder_object(&b,
-				SPA_TYPE_OBJECT_Props, SPA_PARAM_Props,
+				SPA_TYPE_OBJECT_Props,	SPA_PARAM_Props,
 				SPA_PROP_mute,		&SPA_POD_Bool(mute),
 				0));
 	}
diff --git a/src/stream.c b/src/stream.c
index 7ac6e8a7a..d17c0f0d3 100644
--- a/src/stream.c
+++ b/src/stream.c
@@ -202,7 +202,7 @@ static void configure_device(pa_stream *s)
 		else
 			s->device_name = strdup(str);
 	}
-	pw_log_debug("linked to %d '%s'", s->device_index, s->device_name);
+	pw_log_debug("stream %p: linked to %d '%s'", s, s->device_index, s->device_name);
 }
 
 static void stream_state_changed(void *data, enum pw_stream_state old,
@@ -690,6 +690,7 @@ uint32_t pa_stream_get_device_index(pa_stream *s)
 	PA_CHECK_VALIDITY_RETURN_ANY(s->context, s->device_index != PA_INVALID_INDEX,
 			PA_ERR_BADSTATE, PA_INVALID_INDEX);
 
+	pw_log_trace("stream %p: %d", s, s->device_index);
 	return s->device_index;
 }
 
@@ -840,7 +841,7 @@ static int create_stream(pa_stream_direction_t direction,
 		if ((str = getenv("PIPEWIRE_NODE")) != NULL)
 			devid = atoi(str);
 	}
-	else {
+	else if (devid == SPA_ID_INVALID) {
 		uint32_t mask;
 
 		if (direction == PA_STREAM_PLAYBACK)
@@ -1131,7 +1132,7 @@ int pa_stream_peek(pa_stream *s,
 	}
 	*data = SPA_MEMBER(s->buffer_data, s->buffer_offset, void);
 	*nbytes = s->buffer_size;
-	pw_log_trace("stream %p: %p %zd", s, *data, *nbytes);
+	pw_log_trace("stream %p: %p %zd %f", s, *data, *nbytes, *(float*)*data);
 
 	return 0;
 }