stream: somewhat implement drain better

Abuse the xrun callback in the adapter to emit the drained signal until
almost all data left the resampler. This needs more work with a proper
signal and a buffer flag to signal the drain.

spa/plugins/audioconvert/audioadapter.c

@@ -886,11 +886,13 @@ static int impl_node_process(void *object)
 
 			if (status & SPA_STATUS_NEED_DATA) {
 				status = spa_node_process(this->follower);
-				if (!(status & SPA_STATUS_HAVE_DATA))
+				if (!(status & SPA_STATUS_HAVE_DATA)) {
+					spa_node_call_xrun(&this->callbacks, 0, 0, NULL);
 					break;
 				}
 			}
 		}
+	}
 	spa_log_trace_fp(this->log, "%p: process status:%d", this, status);
 
 	this->master = false;

src/modules/module-client-node/remote-node.c

@@ -1131,6 +1131,8 @@ static int node_xrun(void *d, uint64_t trigger, uint64_t delay, struct spa_pod *
 	pw_log_debug("node %p: XRun! count:%u time:%"PRIu64" delay:%"PRIu64" max:%"PRIu64,
 			node, a->xrun_count, trigger, delay, a->max_delay);
 
+	pw_context_driver_emit_xrun(data->context, node);
+
 	return 0;
 }
 

src/pipewire/stream.c

@@ -94,6 +94,7 @@ struct stream {
 	const char *path;
 
 	struct pw_context *context;
+	struct spa_hook context_listener;
 
 	enum spa_direction direction;
 	enum pw_stream_flags flags;
@@ -780,10 +781,6 @@ again:
 			io->buffer_id = SPA_ID_INVALID;
 			io->status = SPA_STATUS_NEED_DATA;
 			pw_log_trace(NAME" %p: no more buffers %p", stream, io);
-			if (impl->draining) {
-				call_drained(impl);
-				goto exit;
-			}
 		}
 	}
 
@@ -795,7 +792,6 @@ again:
 		    io->status == SPA_STATUS_NEED_DATA)
 			goto again;
 	}
-exit:
 	copy_position(impl, impl->queued.outcount);
 
 	res = io->status;
@@ -1039,6 +1035,20 @@ static const struct pw_core_events core_events = {
 	.error = on_core_error,
 };
 
+static void context_xrun(void *data, struct pw_impl_node *node)
+{
+	struct stream *impl = data;
+	if (impl->node != node)
+		return;
+	if (impl->draining)
+		call_drained(impl);
+}
+
+static const struct pw_context_driver_events context_events = {
+	PW_VERSION_CONTEXT_DRIVER_EVENTS,
+	.xrun = context_xrun,
+};
+
 static struct stream *
 stream_new(struct pw_context *context, const char *name,
 		struct pw_properties *props, const struct pw_properties *extra)
@@ -1102,6 +1112,9 @@ stream_new(struct pw_context *context, const char *name,
 	impl->context = context;
 	impl->allow_mlock = context->defaults.mem_allow_mlock;
 
+	spa_hook_list_append(&impl->context->driver_listener_list,
+			&impl->context_listener,
+			&context_events, impl);
 	return impl;
 
 error_properties:
@@ -1226,6 +1239,8 @@ void pw_stream_destroy(struct pw_stream *stream)
 		free(c);
 	}
 
+	spa_hook_remove(&impl->context_listener);
+
 	if (impl->data.context)
 		pw_context_destroy(impl->data.context);
 
@@ -1505,7 +1520,6 @@ pw_stream_connect(struct pw_stream *stream,
 
 	pw_proxy_add_listener(stream->proxy, &stream->proxy_listener, &proxy_events, stream);
 
-
 	pw_impl_node_add_listener(impl->node, &stream->node_listener, &node_events, stream);
 
 	return 0;