node: remove process signals

Remove the process and finish signals, we are not supposed to
call these from rt threads. Instead use the link trigger to signal
the end of the graph.

src/pipewire/node.c

@@ -516,23 +516,6 @@ static void check_properties(struct pw_node *node)
 
 }
 
-static inline int driver_impl_finish(void *data)
-{
-	struct impl *impl = SPA_CONTAINER_OF(data, struct impl, driver_data);
-	struct spa_graph_data *d = &impl->driver_data;
-	struct pw_node *this = &impl->this;
-
-	pw_log_trace("graph %p finish %p", d->graph, impl);
-	pw_node_events_finish(this);
-        return 0;
-}
-
-static const struct spa_graph_callbacks driver_impl_default = {
-        SPA_VERSION_GRAPH_CALLBACKS,
-        .run = spa_graph_impl_run,
-        .finish = driver_impl_finish,
-};
-
 struct pw_node *pw_node_new(struct pw_core *core,
 			    const char *name,
 			    struct pw_properties *properties,
@@ -582,7 +565,7 @@ struct pw_node *pw_node_new(struct pw_core *core,
 	spa_graph_init(&impl->driver_graph, &impl->driver_state);
 	spa_graph_data_init(&impl->driver_data, &impl->driver_graph);
 	spa_graph_set_callbacks(&impl->driver_graph,
-			&driver_impl_default, &impl->driver_data);
+			&spa_graph_impl_default, &impl->driver_data);
 
 	this->rt.driver = &impl->driver_graph;
 	this->rt.activation = &impl->root_activation;
@@ -692,8 +675,6 @@ static void node_process(void *data, int status)
 	pw_log_trace("node %p: process driver:%d exported:%d", node,
 			node->driver, node->exported);
 
-	pw_node_events_process(node);
-
 	if (node->driver && (node->rt.driver->state->pending == 0 || !node->remote)) {
 		struct timespec ts;
 		struct pw_driver_quantum *q = node->rt.quantum;

src/pipewire/node.h

@@ -87,13 +87,6 @@ struct pw_node_events {
 
 	/** the driver of the node changed */
 	void (*driver_changed) (void *data, struct pw_node *driver);
-
-        /** the node wants to process the graph */
-	void (*process) (void *data);
-        /** the node has a buffer to reuse */
-	void (*reuse_buffer) (void *data, uint32_t port_id, uint32_t buffer_id);
-	/** the node driver finished processing */
-	void (*finish) (void *data);
 };
 
 /** Media type of the node, Audio, Video, Midi */

src/pipewire/remote.c

@@ -104,6 +104,12 @@ struct node_data {
 	struct spa_hook proxy_listener;
 
 	struct pw_client_node_position *position;
+
+	struct spa_graph_node_callbacks callbacks;
+        void *callbacks_data;
+
+	struct spa_graph_state state;
+	struct spa_graph_link link;
 };
 
 /** \endcond */
@@ -496,14 +502,6 @@ static void unhandle_socket(struct node_data *data)
                        do_remove_source, 1, NULL, 0, true, data);
 }
 
-static void node_finish(void *data)
-{
-	struct node_data *d = data;
-	uint64_t cmd = 1;
-	pw_log_trace("remote %p: send process", data);
-	write(d->rtwritefd, &cmd, 8);
-}
-
 static void
 on_rtsocket_condition(void *user_data, int fd, enum spa_io mask)
 {
@@ -1309,7 +1307,6 @@ static const struct pw_node_events node_events = {
 	.destroy = node_destroy,
 	.info_changed = node_info_changed,
 	.active_changed = node_active_changed,
-	.finish = node_finish,
 };
 
 static void clear_mix(struct node_data *data, struct mix *mix)
@@ -1345,6 +1342,28 @@ static const struct pw_proxy_events proxy_events = {
 	.destroy = node_proxy_destroy,
 };
 
+static int remote_impl_signal(void *data)
+{
+	struct node_data *d = data;
+	uint64_t cmd = 1;
+	pw_log_trace("remote %p: send process", data);
+	write(d->rtwritefd, &cmd, 8);
+        return 0;
+}
+
+static inline int remote_process(void *data, struct spa_graph_node *node)
+{
+	struct node_data *d = data;
+        spa_debug("remote %p: begin graph", data);
+	spa_graph_state_reset(&d->state);
+	return d->callbacks.process(d->callbacks_data, node);
+}
+
+static const struct spa_graph_node_callbacks impl_root = {
+	SPA_VERSION_GRAPH_NODE_CALLBACKS,
+	.process = remote_process,
+};
+
 struct pw_proxy *pw_remote_export(struct pw_remote *remote,
 				  struct pw_node *node)
 {
@@ -1374,6 +1393,12 @@ struct pw_proxy *pw_remote_export(struct pw_remote *remote,
 	data->core = pw_node_get_core(node);
 	data->node_proxy = (struct pw_client_node_proxy *)proxy;
 
+	data->link.signal = remote_impl_signal;
+	data->link.signal_data = data;
+	data->callbacks = *node->rt.root.callbacks;
+	spa_graph_node_set_callbacks(&node->rt.root, &impl_root, data);
+	spa_graph_link_add(&node->rt.root, &data->state, &data->link);
+
 	node->exported = true;
 
 	spa_list_init(&data->free_mix);