diff --git a/src/context.c b/src/context.c
index 826002b81..56990cde2 100644
--- a/src/context.c
+++ b/src/context.c
@@ -86,6 +86,16 @@ pa_context *pa_context_new(pa_mainloop_api *mainloop, const char *name)
 	return pa_context_new_with_proplist(mainloop, name, NULL);
 }
 
+struct global *pa_context_find_global(pa_context *c, uint32_t id)
+{
+	struct global *g;
+	spa_list_for_each(g, &c->globals, link) {
+		if (g->id == id)
+			return g;
+	}
+	return NULL;
+}
+
 static int set_mask(pa_context *c, struct global *g)
 {
 	const char *str;
@@ -100,10 +110,24 @@ static int set_mask(pa_context *c, struct global *g)
 			g->mask = PA_SUBSCRIPTION_MASK_SINK;
 			g->event = PA_SUBSCRIPTION_EVENT_SINK;
 		}
+		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->dsp_info.session = pa_context_find_global(c,
+					pw_properties_parse_int(str));
+		}
 		else if (strcmp(str, "Audio/Source") == 0) {
 			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, "node.session")) == NULL)
+				return 0;
+			g->mask = PA_SUBSCRIPTION_MASK_DSP_SOURCE;
+			g->dsp_info.session = pa_context_find_global(c,
+					pw_properties_parse_int(str));
+		}
 		else if (strcmp(str, "Stream/Output/Audio") == 0) {
 			g->mask = PA_SUBSCRIPTION_MASK_SINK_INPUT;
 			g->event = PA_SUBSCRIPTION_EVENT_SINK_INPUT;
@@ -112,8 +136,6 @@ static int set_mask(pa_context *c, struct global *g)
 			g->mask = PA_SUBSCRIPTION_MASK_SOURCE_OUTPUT;
 			g->event = PA_SUBSCRIPTION_EVENT_SOURCE_OUTPUT;
 		}
-		else
-			return 0;
 	}
 	else if (g->type == c->t->module) {
 		g->mask = PA_SUBSCRIPTION_MASK_MODULE;
@@ -123,6 +145,23 @@ static int set_mask(pa_context *c, struct global *g)
 		g->mask = PA_SUBSCRIPTION_MASK_CLIENT;
 		g->event = PA_SUBSCRIPTION_EVENT_CLIENT;
 	}
+	else if (g->type == c->t->port) {
+		pw_log_debug("found port %d", g->id);
+	}
+	else if (g->type == c->t->link) {
+                if ((str = pw_properties_get(g->props, "link.output")) == NULL)
+			return 0;
+		g->link_info.src = pa_context_find_global(c, pw_properties_parse_int(str));
+                if ((str = pw_properties_get(g->props, "link.input")) == NULL)
+			return 0;
+		g->link_info.dst = pa_context_find_global(c, pw_properties_parse_int(str));
+
+		if (g->link_info.src == NULL || g->link_info.dst == NULL)
+			return 0;
+
+		pw_log_debug("link %d->%d", g->link_info.src->parent_id,
+				g->link_info.dst->parent_id);
+	}
 	else
 		return 0;
 
@@ -146,6 +185,7 @@ static void registry_event_global(void *data, uint32_t id, uint32_t parent_id,
 	if (set_mask(c, g) == 0)
 		return;
 
+	pw_log_debug("mask %d/%d", g->mask, g->event);
 	spa_list_append(&c->globals, &g->link);
 
 	if (c->subscribe_mask & g->mask) {
@@ -157,16 +197,6 @@ static void registry_event_global(void *data, uint32_t id, uint32_t parent_id,
 	}
 }
 
-struct global *pa_context_find_global(pa_context *c, uint32_t id)
-{
-	struct global *g;
-	spa_list_for_each(g, &c->globals, link) {
-		if (g->id == id)
-			return g;
-	}
-	return NULL;
-}
-
 static void registry_event_global_remove(void *object, uint32_t id)
 {
 	pa_context *c = object;
diff --git a/src/internal.h b/src/internal.h
index c14c6ad74..a478cab63 100644
--- a/src/internal.h
+++ b/src/internal.h
@@ -190,6 +190,10 @@ struct pa_mainloop {
 	int n_events;
 };
 
+#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)
+
 struct global {
 	struct spa_list link;
 	uint32_t id;
@@ -206,6 +210,17 @@ struct global {
 	struct pw_proxy *proxy;
 	struct spa_hook proxy_listener;
         struct spa_hook proxy_proxy_listener;
+
+	/* for links */
+	union {
+		struct {
+			struct global *src;
+			struct global *dst;
+		} link_info;
+		struct {
+			struct global *session;
+		} dsp_info;
+	};
 };
 
 struct pa_context {
@@ -333,6 +348,7 @@ struct pa_stream {
 	struct pw_buffer *dequeued[MAX_BUFFERS];
 	struct spa_ringbuffer dequeued_ring;
 	size_t dequeued_size;
+	size_t maxsize;
 	struct spa_list pending;
 
 	struct pw_buffer *buffer;
diff --git a/src/introspect.c b/src/introspect.c
index cbe9013ed..36af7249d 100644
--- a/src/introspect.c
+++ b/src/introspect.c
@@ -130,6 +130,7 @@ static void sink_callback(struct sink_data *d)
 	spa_zero(i);
 	i.index = g->id;
 	i.name = info->name;
+	i.description = info->name;
 	i.proplist = pa_proplist_new_dict(info->props);
 	i.owner_module = g->parent_id;
 	i.base_volume = PA_VOLUME_NORM;
@@ -339,6 +340,7 @@ static void source_info(pa_operation *o, void *userdata)
 	struct source_data *d = userdata;
 	source_callback(d);
 	d->cb(d->context, NULL, 1, d->userdata);
+	pa_operation_done(o);
 }
 
 pa_operation* pa_context_get_source_info_by_name(pa_context *c, const char *name, pa_source_info_cb_t cb, void *userdata)
@@ -388,6 +390,7 @@ static void source_info_list(pa_operation *o, void *userdata)
 		source_callback(d);
 	}
 	d->cb(c, NULL, 1, d->userdata);
+	pa_operation_done(o);
 }
 
 pa_operation* pa_context_get_source_info_list(pa_context *c, pa_source_info_cb_t cb, void *userdata)
@@ -493,6 +496,7 @@ static void module_info(pa_operation *o, void *userdata)
 	struct module_data *d = userdata;
 	module_callback(d);
 	d->cb(d->context, NULL, 1, d->userdata);
+	pa_operation_done(o);
 }
 
 pa_operation* pa_context_get_module_info(pa_context *c, uint32_t idx, pa_module_info_cb_t cb, void *userdata)
@@ -537,6 +541,7 @@ static void module_info_list(pa_operation *o, void *userdata)
 		module_callback(d);
 	}
 	d->cb(c, NULL, 1, d->userdata);
+	pa_operation_done(o);
 }
 
 pa_operation* pa_context_get_module_info_list(pa_context *c, pa_module_info_cb_t cb, void *userdata)
@@ -601,6 +606,7 @@ static void client_info(pa_operation *o, void *userdata)
 	struct client_data *d = userdata;
 	client_callback(d);
 	d->cb(d->context, NULL, 1, d->userdata);
+	pa_operation_done(o);
 }
 
 pa_operation* pa_context_get_client_info(pa_context *c, uint32_t idx, pa_client_info_cb_t cb, void *userdata)
@@ -645,6 +651,7 @@ static void client_info_list(pa_operation *o, void *userdata)
 		client_callback(d);
 	}
 	d->cb(c, NULL, 1, d->userdata);
+	pa_operation_done(o);
 }
 
 pa_operation* pa_context_get_client_info_list(pa_context *c, pa_client_info_cb_t cb, void *userdata)
@@ -717,14 +724,27 @@ struct sink_input_data {
 	struct global *global;
 };
 
+static pa_stream *find_stream(pa_context *c, uint32_t idx)
+{
+	pa_stream *s;
+	spa_list_for_each(s, &c->streams, link) {
+		if (pw_stream_get_node_id(s->stream) == idx)
+			return s;
+	}
+	return NULL;
+}
+
 static void sink_input_callback(struct sink_input_data *d)
 {
 	struct global *g = d->global;
 	struct pw_node_info *info = g->info;
 	pa_sink_input_info i;
 	pa_format_info ii[1];
+	pa_stream *s;
 
 	pw_log_debug("index %d", g->id);
+	s = find_stream(d->context, g->id);
+
 	spa_zero(i);
 	i.index = g->id;
 	i.name = info->name;
@@ -740,7 +760,11 @@ static void sink_input_callback(struct sink_input_data *d)
 	i.has_volume = true;
 	i.volume_writable = true;
 	i.volume.channels = 1;
-	i.volume.values[0] = PA_VOLUME_NORM;
+	if (s)
+		i.volume.values[0] = s->volume * PA_VOLUME_NORM;
+	else
+		i.volume.values[0] = PA_VOLUME_NORM;
+
 	d->cb(d->context, &i, 0, d->userdata);
 }
 
@@ -749,6 +773,7 @@ static void sink_input_info(pa_operation *o, void *userdata)
 	struct sink_input_data *d = userdata;
 	sink_input_callback(d);
 	d->cb(d->context, NULL, 1, d->userdata);
+	pa_operation_done(o);
 }
 
 pa_operation* pa_context_get_sink_input_info(pa_context *c, uint32_t idx, pa_sink_input_info_cb_t cb, void *userdata)
@@ -792,6 +817,7 @@ static void sink_input_info_list(pa_operation *o, void *userdata)
 		sink_input_callback(d);
 	}
 	d->cb(c, NULL, 1, d->userdata);
+	pa_operation_done(o);
 }
 
 pa_operation* pa_context_get_sink_input_info_list(pa_context *c, pa_sink_input_info_cb_t cb, void *userdata)
@@ -826,16 +852,6 @@ pa_operation* pa_context_move_sink_input_by_index(pa_context *c, uint32_t idx, u
 	return NULL;
 }
 
-static pa_stream *find_stream(pa_context *c, uint32_t idx)
-{
-	pa_stream *s;
-	spa_list_for_each(s, &c->streams, link) {
-		if (pw_stream_get_node_id(s->stream) == idx)
-			return s;
-	}
-	return NULL;
-}
-
 struct success_ack {
 	pa_context_success_cb_t cb;
 	void *userdata;
@@ -845,9 +861,9 @@ static void on_success(pa_operation *o, void *userdata)
 {
 	struct success_ack *d = userdata;
 	pa_context *c = o->context;
-	pa_operation_done(o);
 	if (d->cb)
 		d->cb(c, PA_OK, d->userdata);
+	pa_operation_done(o);
 }
 
 pa_operation* pa_context_set_sink_input_volume(pa_context *c, uint32_t idx, const pa_cvolume *volume, pa_context_success_cb_t cb, void *userdata)
diff --git a/src/stream.c b/src/stream.c
index 1a7dee4a6..6a6af3ee2 100644
--- a/src/stream.c
+++ b/src/stream.c
@@ -96,10 +96,70 @@ static int dequeue_buffer(pa_stream *s)
 	return 0;
 }
 
+static void dump_buffer_attr(pa_stream *s, pa_buffer_attr *attr)
+{
+	pw_log_info("stream %p: maxlength: %u", s, attr->maxlength);
+	pw_log_info("stream %p: tlength: %u", s, attr->tlength);
+	pw_log_info("stream %p: minreq: %u", s, attr->minreq);
+	pw_log_info("stream %p: prebuf: %u", s, attr->prebuf);
+	pw_log_info("stream %p: fragsize: %u", s, attr->fragsize);
+}
+
 static void configure_buffers(pa_stream *s)
 {
-	s->buffer_attr.maxlength = 65536;
+	s->buffer_attr.maxlength = s->maxsize;
 	s->buffer_attr.prebuf = s->buffer_attr.minreq;
+	s->buffer_attr.fragsize = s->buffer_attr.minreq;
+	dump_buffer_attr(s, &s->buffer_attr);
+}
+
+static struct global *find_linked(pa_stream *s, uint32_t idx)
+{
+	struct global *g, *f;
+	pa_context *c = s->context;
+
+	spa_list_for_each(g, &c->globals, link) {
+		if (g->type != c->t->link)
+			continue;
+
+		pw_log_debug("%d %d %d", idx,
+				g->link_info.src->parent_id,
+				g->link_info.dst->parent_id);
+
+		if (g->link_info.src->parent_id == idx)
+			f = pa_context_find_global(c, g->link_info.dst->parent_id);
+		else if (g->link_info.dst->parent_id == 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) {
+			f = f->dsp_info.session;
+		}
+		return f;
+	}
+	return NULL;
+}
+static void configure_device(pa_stream *s)
+{
+	struct global *g;
+	const char *str;
+
+	g = find_linked(s, pa_stream_get_index(s));
+	if (g == NULL) {
+		s->device_index = PA_INVALID_INDEX;
+		s->device_name = NULL;
+	}
+	else {
+		s->device_index = g->id;
+		if ((str = pw_properties_get(g->props, "node.name")) == NULL)
+			s->device_name = strdup("unknown");
+		else
+			s->device_name = strdup(str);
+	}
+	pw_log_debug("linked to %d '%s'", s->device_index, s->device_name);
 }
 
 static void stream_state_changed(void *data, enum pw_stream_state old,
@@ -122,6 +182,7 @@ static void stream_state_changed(void *data, enum pw_stream_state old,
 	case PW_STREAM_STATE_READY:
 		break;
 	case PW_STREAM_STATE_PAUSED:
+		configure_device(s);
 		configure_buffers(s);
 		pa_stream_set_state(s, PA_STREAM_READY);
 		break;
@@ -219,14 +280,9 @@ static void patch_buffer_attr(pa_stream *s, pa_buffer_attr *attr, pa_stream_flag
 	if (attr->fragsize == (uint32_t) -1)
 		attr->fragsize = attr->tlength; /* Pass data to the app only when the buffer is filled up once */
 
-	pw_log_info("stream %p: maxlength: %u", s, attr->maxlength);
-	pw_log_info("stream %p: tlength: %u", s, attr->tlength);
-	pw_log_info("stream %p: minreq: %u", s, attr->minreq);
-	pw_log_info("stream %p: prebuf: %u", s, attr->prebuf);
-	pw_log_info("stream %p: fragsize: %u", s, attr->fragsize);
+	dump_buffer_attr(s, attr);
 }
 
-
 static void stream_format_changed(void *data, const struct spa_pod *format)
 {
 	pa_stream *s = data;
@@ -271,6 +327,17 @@ static void stream_format_changed(void *data, const struct spa_pod *format)
 	pw_stream_finish_format(s->stream, res, params, n_params);
 }
 
+static void stream_add_buffer(void *data, struct pw_buffer *buffer)
+{
+	pa_stream *s = data;
+	s->maxsize += buffer->buffer->datas[0].maxsize;
+}
+static void stream_remove_buffer(void *data, struct pw_buffer *buffer)
+{
+	pa_stream *s = data;
+	s->maxsize -= buffer->buffer->datas[0].maxsize;
+}
+
 static void stream_process(void *data)
 {
 	pa_stream *s = data;
@@ -295,6 +362,8 @@ static const struct pw_stream_events stream_events =
 	PW_VERSION_STREAM_EVENTS,
 	.state_changed = stream_state_changed,
 	.format_changed = stream_format_changed,
+	.add_buffer = stream_add_buffer,
+	.remove_buffer = stream_remove_buffer,
 	.process = stream_process,
 };
 
@@ -378,9 +447,7 @@ pa_stream* stream_new(pa_context *c, const char *name,
 	s->buffer_attr.fragsize = (uint32_t) -1;
 
 	s->device_index = PA_INVALID_INDEX;
-
-	s->device_index = 0;
-	s->device_name = strdup("unknown");
+	s->device_name = NULL;
 
 	spa_ringbuffer_init(&s->dequeued_ring);
 
@@ -479,6 +546,7 @@ void pa_stream_set_state(pa_stream *s, pa_stream_state_t st) {
 
 	pa_stream_ref(s);
 
+	pw_log_debug("stream %p: state %d -> %d", s, s->state, st);
 	s->state = st;
 
 	if (s->state_callback)
@@ -1279,7 +1347,7 @@ int pa_stream_get_time(pa_stream *s, pa_usec_t *r_usec)
 	if (r_usec)
 		*r_usec = res;
 
-	pw_log_debug("stream %p: %ld %ld %ld %ld %d/%d %ld", s, now, t.now, delay, t.ticks, t.rate.num, t.rate.denom, res);
+	pw_log_trace("stream %p: %ld %ld %ld %ld %d/%d %ld", s, now, t.now, delay, t.ticks, t.rate.num, t.rate.denom, res);
 
 	return 0;
 }