diff --git a/src/examples/export-sink.c b/src/examples/export-sink.c
index 952a7b406..7d0d6e44b 100644
--- a/src/examples/export-sink.c
+++ b/src/examples/export-sink.c
@@ -441,9 +441,10 @@ static int impl_port_use_buffers(struct spa_node *node, enum spa_direction direc
 	return 0;
 }
 
-static int impl_node_process_input(struct spa_node *node)
+static int do_render(struct spa_loop *loop, bool async, uint32_t seq,
+		     const void *_data, size_t size, void *user_data)
 {
-	struct data *d = SPA_CONTAINER_OF(node, struct data, impl_node);
+	struct data *d = user_data;
 	struct spa_buffer *buf;
 	uint8_t *map;
 	void *sdata, *ddata;
@@ -452,10 +453,10 @@ static int impl_node_process_input(struct spa_node *node)
 	uint8_t *src, *dst;
 
 	if (d->io->status != SPA_STATUS_HAVE_BUFFER)
-		goto done;
+		return 0;
 
 	if (d->io->buffer_id > d->n_buffers)
-		goto done;
+		return 0;
 
 	buf = d->buffers[d->io->buffer_id];
 
@@ -493,7 +494,18 @@ static int impl_node_process_input(struct spa_node *node)
 	if (map)
 		munmap(map, buf->datas[0].maxsize + buf->datas[0].mapoffset);
 
-done:
+	return 0;
+}
+
+static int impl_node_process_input(struct spa_node *node)
+{
+	struct data *d = SPA_CONTAINER_OF(node, struct data, impl_node);
+	int res;
+
+	if ((res = pw_loop_invoke(pw_main_loop_get_loop(d->loop), do_render,
+				  SPA_ID_INVALID, NULL, 0, true, d)) < 0)
+		return res;
+
 	handle_events(d);
 
 	return d->io->status = SPA_STATUS_NEED_BUFFER;
diff --git a/src/examples/local-v4l2.c b/src/examples/local-v4l2.c
index fab7ee5e2..b71adae9f 100644
--- a/src/examples/local-v4l2.c
+++ b/src/examples/local-v4l2.c
@@ -390,9 +390,10 @@ static int impl_port_use_buffers(struct spa_node *node, enum spa_direction direc
 	return 0;
 }
 
-static int impl_node_process_input(struct spa_node *node)
+static int do_render(struct spa_loop *loop, bool async, uint32_t seq,
+		     const void *_data, size_t size, void *user_data)
 {
-	struct data *d = SPA_CONTAINER_OF(node, struct data, impl_node);
+	struct data *d = user_data;
 	struct spa_buffer *buf;
 	uint8_t *map;
 	void *sdata, *ddata;
@@ -436,6 +437,18 @@ static int impl_node_process_input(struct spa_node *node)
 	if (map)
 		munmap(map, buf->datas[0].maxsize + buf->datas[0].mapoffset);
 
+	return 0;
+}
+
+static int impl_node_process_input(struct spa_node *node)
+{
+	struct data *d = SPA_CONTAINER_OF(node, struct data, impl_node);
+	int res;
+
+	if ((res = pw_loop_invoke(pw_main_loop_get_loop(d->loop), do_render,
+				  SPA_ID_INVALID, NULL, 0, true, d)) < 0)
+		return res;
+
 	handle_events(d);
 
 	d->io->status = SPA_STATUS_NEED_BUFFER;
diff --git a/src/examples/video-play.c b/src/examples/video-play.c
index c8dc8e077..4059a62bc 100644
--- a/src/examples/video-play.c
+++ b/src/examples/video-play.c
@@ -96,20 +96,18 @@ static void handle_events(struct data *data)
 	}
 }
 
-static void
-on_stream_new_buffer(void *_data, uint32_t id)
+static int
+do_render(struct spa_loop *loop, bool async, uint32_t seq,
+	  const void *_data, size_t size, void *user_data)
 {
-	struct data *data = _data;
-	struct pw_stream *stream = data->stream;
-	struct spa_buffer *buf;
+	struct data *data = user_data;
+	struct spa_buffer *buf = ((struct spa_buffer **) _data)[0];
 	uint8_t *map;
 	void *sdata, *ddata;
 	int sstride, dstride, ostride;
 	int i;
 	uint8_t *src, *dst;
 
-	buf = pw_stream_peek_buffer(stream, id);
-
 	if (buf->datas[0].type == data->type.data.MemFd ||
 	    buf->datas[0].type == data->type.data.DmaBuf) {
 		map = mmap(NULL, buf->datas[0].maxsize + buf->datas[0].mapoffset, PROT_READ,
@@ -119,11 +117,11 @@ on_stream_new_buffer(void *_data, uint32_t id)
 		map = NULL;
 		sdata = buf->datas[0].data;
 	} else
-		return;
+		return -EINVAL;
 
 	if (SDL_LockTexture(data->texture, NULL, &ddata, &dstride) < 0) {
 		fprintf(stderr, "Couldn't lock texture: %s\n", SDL_GetError());
-		return;
+		return -EIO;
 	}
 	sstride = buf->datas[0].chunk->stride;
 	ostride = SPA_MIN(sstride, dstride);
@@ -144,6 +142,22 @@ on_stream_new_buffer(void *_data, uint32_t id)
 	if (map)
 		munmap(map, buf->datas[0].maxsize + buf->datas[0].mapoffset);
 
+	return 0;
+}
+
+static void
+on_stream_new_buffer(void *_data, uint32_t id)
+{
+	struct data *data = _data;
+	struct pw_stream *stream = data->stream;
+	struct spa_buffer *buf;
+
+	buf = pw_stream_peek_buffer(stream, id);
+
+	pw_loop_invoke(pw_main_loop_get_loop(data->loop), do_render,
+		       SPA_ID_INVALID, &buf, sizeof(struct spa_buffer *),
+		       true, data);
+
 	pw_stream_recycle_buffer(stream, id);
 
 	handle_events(data);