interfaces: improve remote API

Add return values to events and method callbacks. This makes it
possible to pass any error or async return value.
Add sync/done/error in both directions so that both proxy and resource
and perform/reply sync and produce errors.
Return a SPA_ASYNC from remote method calls (and events), keep the
sequence number in the connection.
With the core sync/done we can remove the client-node done method and
it's internal sequence number along with the seq number in method calls.
We can also use the method/event async return value to perform a sync
with as the unique sequence number for this method.
Add sync method and done/error event to proxy and resource.

spa/examples/example-control.c

@@ -184,16 +184,6 @@ static int make_node(struct data *data, struct spa_node **node, const char *lib,
 	return -EBADF;
 }
 
-static void on_sink_done(void *data, int seq, int res)
-{
-	printf("got done %d %d\n", seq, res);
-}
-
-static void on_sink_event(void *data, struct spa_event *event)
-{
-	printf("got event %d\n", SPA_EVENT_TYPE(event));
-}
-
 static void update_props(struct data *data)
 {
 	struct spa_pod_builder b;
@@ -233,7 +223,7 @@ static void update_props(struct data *data)
 		data->volume_accum -= M_PI_M2;
 }
 
-static void on_sink_ready(void *_data, int status)
+static int on_sink_ready(void *_data, int status)
 {
 	struct data *data = _data;
 
@@ -241,20 +231,20 @@ static void on_sink_ready(void *_data, int status)
 
 	spa_graph_node_process(&data->source_node);
 	spa_graph_node_process(&data->sink_node);
+	return 0;
 }
 
-static void
+static int
 on_sink_reuse_buffer(void *_data, uint32_t port_id, uint32_t buffer_id)
 {
 	struct data *data = _data;
 
 	data->source_sink_io[0].buffer_id = buffer_id;
+	return 0;
 }
 
 static const struct spa_node_callbacks sink_callbacks = {
 	SPA_VERSION_NODE_CALLBACKS,
-	.done = on_sink_done,
-	.event = on_sink_event,
 	.ready = on_sink_ready,
 	.reuse_buffer = on_sink_reuse_buffer
 };

spa/examples/local-v4l2.c

@@ -145,21 +145,7 @@ static void handle_events(struct data *data)
 	}
 }
 
-static void on_source_done(void *data, int seq, int res)
-{
-	printf("got done %d %d\n", seq, res);
-}
-
-static void on_source_event(void *_data, struct spa_event *event)
-{
-	struct data *data = _data;
-
-	handle_events(data);
-
-	printf("got event %d\n", SPA_EVENT_TYPE(event));
-}
-
-static void on_source_ready(void *_data, int status)
+static int on_source_ready(void *_data, int status)
 {
 	struct data *data = _data;
 	int res;
@@ -177,7 +163,7 @@ static void on_source_ready(void *_data, int status)
 		printf("got process error %d\n", res);
 
 	if (io->buffer_id > MAX_BUFFERS)
-		return;
+		return -EINVAL;
 
 	b = &data->buffers[io->buffer_id];
 
@@ -194,26 +180,28 @@ static void on_source_ready(void *_data, int status)
 
 		if (SDL_LockTexture(texture, NULL, &sdata, &sstride) < 0) {
 			fprintf(stderr, "Couldn't lock texture: %s\n", SDL_GetError());
-			return;
+			return -EIO;
 		}
 	} else {
 		uint8_t *map;
 
 		if (SDL_LockTexture(data->texture, NULL, &ddata, &dstride) < 0) {
 			fprintf(stderr, "Couldn't lock texture: %s\n", SDL_GetError());
-			return;
+			return -EIO;
 		}
 		sdata = datas[0].data;
 		if (datas[0].type == SPA_DATA_MemFd ||
 		    datas[0].type == SPA_DATA_DmaBuf) {
 			map = mmap(NULL, datas[0].maxsize + datas[0].mapoffset, PROT_READ,
 				   MAP_PRIVATE, datas[0].fd, 0);
+			if (map == MAP_FAILED)
+				return -errno;
 			sdata = SPA_MEMBER(map, datas[0].mapoffset, uint8_t);
 		} else if (datas[0].type == SPA_DATA_MemPtr) {
 			map = NULL;
 			sdata = datas[0].data;
 		} else
-			return;
+			return -EIO;
 
 		sstride = datas[0].chunk->stride;
 
@@ -233,12 +221,11 @@ static void on_source_ready(void *_data, int status)
 	}
 
 	io->status = SPA_STATUS_NEED_BUFFER;
+	return 0;
 }
 
 static const struct spa_node_callbacks source_callbacks = {
 	SPA_VERSION_NODE_CALLBACKS,
-	.done = on_source_done,
-	.event = on_source_event,
 	.ready = on_source_ready
 };
 

spa/include/spa/monitor/device.h

@@ -74,15 +74,15 @@ struct spa_device_callbacks {
 	uint32_t version;
 
 	/** notify extra information about the device */
-	void (*info) (void *data, const struct spa_device_info *info);
+	int (*info) (void *data, const struct spa_device_info *info);
 
 	/** a device event */
-	void (*event) (void *data, struct spa_event *event);
+	int (*event) (void *data, struct spa_event *event);
 
 	/** info changed for an object managed by the device, info is NULL when
 	 * the object is removed */
-	void (*object_info) (void *data, uint32_t id,
-			     const struct spa_device_object_info *info);
+	int (*object_info) (void *data, uint32_t id,
+		const struct spa_device_object_info *info);
 };
 
 /**

spa/include/spa/monitor/monitor.h

@@ -80,10 +80,10 @@ struct spa_monitor_callbacks {
 	uint32_t version;
 
 	/** receive extra information about the monitor */
-	void (*info) (void *data, const struct spa_dict *info);
+	int (*info) (void *data, const struct spa_dict *info);
 
 	/** an item is added/removed/changed on the monitor */
-	void (*event) (void *data, struct spa_event *event);
+	int (*event) (void *data, struct spa_event *event);
 };
 
 /**

spa/include/spa/node/node.h

@@ -97,18 +97,22 @@ struct spa_node_callbacks {
 #define SPA_VERSION_NODE_CALLBACKS	0
 	uint32_t version;	/**< version of this structure */
 
-	/** Emited when info changes */
-	void (*info) (void *data, const struct spa_node_info *info);
-
-	/** Emited when port info changes, NULL when port is removed */
-	void (*port_info) (void *data,
-		           enum spa_direction direction, uint32_t port,
-			   const struct spa_port_info *info);
-
-	/** Emited when an async operation completed.
+	/** Emited as a reply to a sync method with \a seq.
 	 *
 	 * Will be called from the main thread. */
-	void (*done) (void *data, int seq, int res);
+	int (*done) (void *data, uint32_t seq);
+
+	/** an asynchronous error occured */
+	int (*error) (void *data, int res, const char *message);
+
+	/** Emited when info changes */
+	int (*info) (void *data, const struct spa_node_info *info);
+
+	/** Emited when port info changes, NULL when port is removed */
+	int (*port_info) (void *data,
+			enum spa_direction direction, uint32_t port,
+			const struct spa_port_info *info);
+
 	/**
 	 * \param node a spa_node
 	 * \param event the event that was emited
@@ -116,7 +120,7 @@ struct spa_node_callbacks {
 	 * This will be called when an out-of-bound event is notified
 	 * on \a node. The callback will be called from the main thread.
 	 */
-	void (*event) (void *data, struct spa_event *event);
+	int (*event) (void *data, struct spa_event *event);
 
 	/**
 	 * \param node a spa_node
@@ -127,7 +131,7 @@ struct spa_node_callbacks {
 	 * When this function is NULL, synchronous operation is requested
 	 * on the ports.
 	 */
-	void (*ready) (void *data, int state);
+	int (*ready) (void *data, int state);
 
 	/**
 	 * \param node a spa_node
@@ -140,9 +144,9 @@ struct spa_node_callbacks {
 	 * When this function is NULL, the buffers to reuse will be set in
 	 * the io area of the input ports.
 	 */
-	void (*reuse_buffer) (void *data,
-			      uint32_t port_id,
-			      uint32_t buffer_id);
+	int (*reuse_buffer) (void *data,
+			     uint32_t port_id,
+			     uint32_t buffer_id);
 
 };
 
@@ -160,6 +164,32 @@ struct spa_node {
 	 * structure in the future */
 #define SPA_VERSION_NODE	0
 	uint32_t version;
+
+	/**
+	 * Set callbacks to receive events and scheduling callbacks from \a node.
+	 * if \a callbacks is NULL, the current callbacks are removed.
+	 *
+	 * This function must be called from the main thread.
+	 *
+	 * \param node a spa_node
+	 * \param callbacks callbacks to set
+	 * \return 0 on success
+	 *         -EINVAL when node is NULL
+	 */
+	int (*set_callbacks) (struct spa_node *node,
+			      const struct spa_node_callbacks *callbacks,
+			      void *data);
+	/**
+	 * Perform a sync operation.
+	 *
+	 * Calling this method will emit the done event or -EIO when
+	 * no callbacks are installed.
+	 *
+	 * Because all methods are serialized in the node, this can be used
+	 * to wait for completion of all previous method calls.
+	 */
+	int (*sync) (struct spa_node *node, uint32_t seq);
+
 	/**
 	 * Enumerate the parameters of a node.
 	 *
@@ -251,20 +281,6 @@ struct spa_node {
 	 */
 	int (*send_command) (struct spa_node *node, const struct spa_command *command);
 
-	/**
-	 * Set callbacks to receive events and scheduling callbacks from \a node.
-	 * if \a callbacks is NULL, the current callbacks are removed.
-	 *
-	 * This function must be called from the main thread.
-	 *
-	 * \param node a spa_node
-	 * \param callbacks callbacks to set
-	 * \return 0 on success
-	 *         -EINVAL when node is NULL
-	 */
-	int (*set_callbacks) (struct spa_node *node,
-			      const struct spa_node_callbacks *callbacks,
-			      void *data);
 	/**
 	 * Make a new port with \a port_id. The caller should use get_port_ids() to
 	 * find an unused id for the given \a direction.
@@ -477,11 +493,12 @@ struct spa_node {
 	int (*process) (struct spa_node *node);
 };
 
+#define spa_node_set_callbacks(n,...)		(n)->set_callbacks((n),__VA_ARGS__)
+#define spa_node_sync(n,...)			(n)->sync((n),__VA_ARGS__)
 #define spa_node_enum_params(n,...)		(n)->enum_params((n),__VA_ARGS__)
 #define spa_node_set_param(n,...)		(n)->set_param((n),__VA_ARGS__)
 #define spa_node_set_io(n,...)			(n)->set_io((n),__VA_ARGS__)
 #define spa_node_send_command(n,...)		(n)->send_command((n),__VA_ARGS__)
-#define spa_node_set_callbacks(n,...)		(n)->set_callbacks((n),__VA_ARGS__)
 #define spa_node_add_port(n,...)		(n)->add_port((n),__VA_ARGS__)
 #define spa_node_remove_port(n,...)		(n)->remove_port((n),__VA_ARGS__)
 #define spa_node_port_enum_params(n,...)	(n)->port_enum_params((n),__VA_ARGS__)

spa/plugins/alsa/alsa-sink.c

@@ -264,6 +264,20 @@ static void emit_port_info(struct state *this)
 	}
 }
 
+static int
+impl_node_sync(struct spa_node *node, uint32_t seq)
+{
+	struct state *this;
+
+	spa_return_val_if_fail(node != NULL, -EINVAL);
+	this = SPA_CONTAINER_OF(node, struct state, node);
+
+	spa_return_val_if_fail(this->callbacks && this->callbacks->done, -EIO);
+
+	this->callbacks->done(this->callbacks_data, seq);
+	return 0;
+}
+
 static int
 impl_node_set_callbacks(struct spa_node *node,
 			const struct spa_node_callbacks *callbacks,
@@ -637,11 +651,12 @@ static int impl_node_process(struct spa_node *node)
 
 static const struct spa_node impl_node = {
 	SPA_VERSION_NODE,
+	.set_callbacks = impl_node_set_callbacks,
+	.sync = impl_node_sync,
 	.enum_params = impl_node_enum_params,
 	.set_param = impl_node_set_param,
 	.set_io = impl_node_set_io,
 	.send_command = impl_node_send_command,
-	.set_callbacks = impl_node_set_callbacks,
 	.add_port = impl_node_add_port,
 	.remove_port = impl_node_remove_port,
 	.port_enum_params = impl_node_port_enum_params,

spa/plugins/audioconvert/audioconvert.c

@@ -550,16 +550,17 @@ static void emit_port_info(struct impl *this,
 }
 
 
-static void fmt_input_port_info(void *data,
+static int fmt_input_port_info(void *data,
 		enum spa_direction direction, uint32_t port,
 		const struct spa_port_info *info)
 {
 	struct impl *this = data;
 
 	if (direction != SPA_DIRECTION_INPUT)
-		return;
+		return 0;
 
 	emit_port_info(this, direction, port, info);
+	return 0;
 }
 
 static struct spa_node_callbacks fmt_input_callbacks = {
@@ -567,16 +568,17 @@ static struct spa_node_callbacks fmt_input_callbacks = {
 	.port_info = fmt_input_port_info
 };
 
-static void fmt_output_port_info(void *data,
+static int fmt_output_port_info(void *data,
 		enum spa_direction direction, uint32_t port,
 		const struct spa_port_info *info)
 {
 	struct impl *this = data;
 
 	if (direction != SPA_DIRECTION_OUTPUT)
-		return;
+		return 0;
 
 	emit_port_info(this, direction, port, info);
+	return 0;
 }
 
 static struct spa_node_callbacks fmt_output_callbacks = {

spa/tools/spa-inspect.c

@@ -136,7 +136,7 @@ inspect_port_params(struct data *data, struct spa_node *node,
 	}
 }
 
-static void node_info(void *_data, const struct spa_node_info *info)
+static int node_info(void *_data, const struct spa_node_info *info)
 {
 	struct data *data = _data;
 
@@ -148,9 +148,10 @@ static void node_info(void *_data, const struct spa_node_info *info)
 		spa_debug_dict(2, info->props);
 	}
 	inspect_node_params(data, data->node);
+	return 0;
 }
 
-static void node_port_info(void *_data, enum spa_direction direction, uint32_t id,
+static int node_port_info(void *_data, enum spa_direction direction, uint32_t id,
 		const struct spa_port_info *info)
 {
 	struct data *data = _data;
@@ -164,6 +165,7 @@ static void node_port_info(void *_data, enum spa_direction direction, uint32_t i
 				id);
 		inspect_port_params(data, data->node, direction, id);
 	}
+	return 0;
 }
 
 static const struct spa_node_callbacks node_callbacks =

spa/tools/spa-monitor.c

@@ -61,12 +61,13 @@ static void inspect_item(struct data *data, struct spa_pod *item)
 	spa_debug_pod(0, NULL, item);
 }
 
-static void on_monitor_info(void *_data, const struct spa_dict *info)
+static int on_monitor_info(void *_data, const struct spa_dict *info)
 {
 	spa_debug_dict(0, info);
+	return 0;
 }
 
-static void on_monitor_event(void *_data, struct spa_event *event)
+static int on_monitor_event(void *_data, struct spa_event *event)
 {
 	struct data *data = _data;
 
@@ -84,6 +85,7 @@ static void on_monitor_event(void *_data, struct spa_event *event)
 		inspect_item(data, SPA_POD_CONTENTS(struct spa_event, event));
 		break;
 	}
+	return 0;
 }
 
 static int do_add_source(struct spa_loop *loop, struct spa_source *source)