1 files changed, 428 insertions, 0 deletions
diff --git a/src/modules/module-protocol-pulse/stream.c b/src/modules/module-protocol-pulse/stream.c
new file mode 100644
index 0000000..59fb8a3
--- /dev/null
+++ b/src/modules/module-protocol-pulse/stream.c
@@ -0,0 +1,428 @@
+/* PipeWire
+ *
+ * Copyright © 2020 Wim Taymans
+ *
+ * Permission is hereby granted, free of charge, to any person obtaining a
+ * copy of this software and associated documentation files (the "Software"),
+ * to deal in the Software without restriction, including without limitation
+ * the rights to use, copy, modify, merge, publish, distribute, sublicense,
+ * and/or sell copies of the Software, and to permit persons to whom the
+ * Software is furnished to do so, subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice (including the next
+ * paragraph) shall be included in all copies or substantial portions of the
+ * Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+ * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT.  IN NO EVENT SHALL
+ * THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+ * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING
+ * FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER
+ * DEALINGS IN THE SOFTWARE.
+ */
+
+#include <stdbool.h>
+#include <stdint.h>
+#include <stdlib.h>
+
+#include <spa/utils/hook.h>
+#include <spa/utils/ringbuffer.h>
+#include <pipewire/log.h>
+#include <pipewire/loop.h>
+#include <pipewire/map.h>
+#include <pipewire/properties.h>
+#include <pipewire/stream.h>
+#include <pipewire/work-queue.h>
+
+#include "client.h"
+#include "commands.h"
+#include "defs.h"
+#include "internal.h"
+#include "log.h"
+#include "message.h"
+#include "reply.h"
+#include "stream.h"
+
+static int parse_frac(struct pw_properties *props, const char *key,
+		const struct spa_fraction *def, struct spa_fraction *res)
+{
+	const char *str;
+	if (props == NULL ||
+	    (str = pw_properties_get(props, key)) == NULL ||
+	    sscanf(str, "%u/%u", &res->num, &res->denom) != 2 ||
+	     res->denom == 0) {
+		*res = *def;
+	}
+	return 0;
+}
+
+struct stream *stream_new(struct client *client, enum stream_type type, uint32_t create_tag,
+			  const struct sample_spec *ss, const struct channel_map *map,
+			  const struct buffer_attr *attr)
+{
+	int res;
+	struct defs *defs = &client->impl->defs;
+	const char *str;
+
+	struct stream *stream = calloc(1, sizeof(*stream));
+	if (stream == NULL)
+		return NULL;
+
+	stream->channel = pw_map_insert_new(&client->streams, stream);
+	if (stream->channel == SPA_ID_INVALID)
+		goto error_errno;
+
+	stream->impl = client->impl;
+	stream->client = client;
+	stream->type = type;
+	stream->create_tag = create_tag;
+	stream->ss = *ss;
+	stream->map = *map;
+	stream->attr = *attr;
+	spa_ringbuffer_init(&stream->ring);
+
+	stream->peer_index = SPA_ID_INVALID;
+
+	parse_frac(client->props, "pulse.min.req", &defs->min_req, &stream->min_req);
+	parse_frac(client->props, "pulse.min.frag", &defs->min_frag, &stream->min_frag);
+	parse_frac(client->props, "pulse.min.quantum", &defs->min_quantum, &stream->min_quantum);
+	parse_frac(client->props, "pulse.default.req", &defs->default_req, &stream->default_req);
+	parse_frac(client->props, "pulse.default.frag", &defs->default_frag, &stream->default_frag);
+	parse_frac(client->props, "pulse.default.tlength", &defs->default_tlength, &stream->default_tlength);
+	stream->idle_timeout_sec = defs->idle_timeout;
+	if ((str = pw_properties_get(client->props, "pulse.idle.timeout")) != NULL)
+		spa_atou32(str, &stream->idle_timeout_sec, 0);
+
+	switch (type) {
+	case STREAM_TYPE_RECORD:
+		stream->direction = PW_DIRECTION_INPUT;
+		break;
+	case STREAM_TYPE_PLAYBACK:
+	case STREAM_TYPE_UPLOAD:
+		stream->direction = PW_DIRECTION_OUTPUT;
+		break;
+	default:
+		spa_assert_not_reached();
+	}
+
+	return stream;
+
+error_errno:
+	res = errno;
+	free(stream);
+	errno = res;
+
+	return NULL;
+}
+
+void stream_free(struct stream *stream)
+{
+	struct client *client = stream->client;
+	struct impl *impl = client->impl;
+
+	pw_log_debug("client %p: stream %p channel:%d", client, stream, stream->channel);
+
+	if (stream->pending)
+		spa_list_remove(&stream->link);
+
+	if (stream->drain_tag)
+		reply_error(client, -1, stream->drain_tag, -ENOENT);
+
+	if (stream->killed)
+		stream_send_killed(stream);
+
+	if (stream->stream) {
+		spa_hook_remove(&stream->stream_listener);
+		pw_stream_disconnect(stream->stream);
+
+		/* force processing of all pending messages before we destroy
+		 * the stream */
+		pw_loop_invoke(impl->loop, NULL, 0, NULL, 0, false, client);
+
+		pw_stream_destroy(stream->stream);
+	}
+	if (stream->channel != SPA_ID_INVALID)
+		pw_map_remove(&client->streams, stream->channel);
+
+	pw_work_queue_cancel(impl->work_queue, stream, SPA_ID_INVALID);
+
+	if (stream->buffer)
+		free(stream->buffer);
+
+	pw_properties_free(stream->props);
+
+	free(stream);
+}
+
+void stream_flush(struct stream *stream)
+{
+	pw_stream_flush(stream->stream, false);
+
+	if (stream->type == STREAM_TYPE_PLAYBACK) {
+		stream->ring.writeindex = stream->ring.readindex;
+		stream->write_index = stream->read_index;
+
+		if (stream->attr.prebuf > 0)
+			stream->in_prebuf = true;
+
+		stream->playing_for = 0;
+		stream->underrun_for = -1;
+		stream->is_underrun = true;
+
+		stream_send_request(stream);
+	} else {
+		stream->ring.readindex = stream->ring.writeindex;
+		stream->read_index = stream->write_index;
+	}
+}
+
+static bool stream_prebuf_active(struct stream *stream, int32_t avail)
+{
+	if (stream->in_prebuf) {
+		if (avail >= (int32_t) stream->attr.prebuf)
+			stream->in_prebuf = false;
+	} else {
+		if (stream->attr.prebuf > 0 && avail <= 0)
+			stream->in_prebuf = true;
+	}
+	return stream->in_prebuf;
+}
+
+uint32_t stream_pop_missing(struct stream *stream)
+{
+	int64_t missing, avail;
+
+	avail = stream->write_index - stream->read_index;
+
+	missing = stream->attr.tlength;
+	missing -= stream->requested;
+	missing -= avail;
+
+	if (missing <= 0)
+		return 0;
+
+	if (missing < stream->attr.minreq && !stream_prebuf_active(stream, avail))
+		return 0;
+
+	stream->requested += missing;
+
+	return missing;
+}
+
+void stream_set_paused(struct stream *stream, bool paused, const char *reason)
+{
+	if (stream->is_paused == paused)
+		return;
+
+	if (reason && stream->client)
+		pw_log_info("%p: [%s] %s because of %s",
+				stream, stream->client->name,
+				paused ? "paused" : "resumed", reason);
+
+	stream->is_paused = paused;
+	pw_stream_set_active(stream->stream, !paused);
+}
+
+int stream_send_underflow(struct stream *stream, int64_t offset)
+{
+	struct client *client = stream->client;
+	struct impl *impl = client->impl;
+	struct message *reply;
+
+	if (ratelimit_test(&impl->rate_limit, stream->timestamp, SPA_LOG_LEVEL_INFO)) {
+		pw_log_info("[%s]: UNDERFLOW channel:%u offset:%" PRIi64,
+			    client->name, stream->channel, offset);
+	}
+
+	reply = message_alloc(impl, -1, 0);
+	message_put(reply,
+		TAG_U32, COMMAND_UNDERFLOW,
+		TAG_U32, -1,
+		TAG_U32, stream->channel,
+		TAG_INVALID);
+
+	if (client->version >= 23) {
+		message_put(reply,
+			TAG_S64, offset,
+			TAG_INVALID);
+	}
+
+	return client_queue_message(client, reply);
+}
+
+int stream_send_overflow(struct stream *stream)
+{
+	struct client *client = stream->client;
+	struct impl *impl = client->impl;
+	struct message *reply;
+
+	pw_log_warn("client %p [%s]: stream %p OVERFLOW channel:%u",
+		    client, client->name, stream, stream->channel);
+
+	reply = message_alloc(impl, -1, 0);
+	message_put(reply,
+		TAG_U32, COMMAND_OVERFLOW,
+		TAG_U32, -1,
+		TAG_U32, stream->channel,
+		TAG_INVALID);
+
+	return client_queue_message(client, reply);
+}
+
+int stream_send_killed(struct stream *stream)
+{
+	struct client *client = stream->client;
+	struct impl *impl = client->impl;
+	struct message *reply;
+	uint32_t command;
+
+	command = stream->direction == PW_DIRECTION_OUTPUT ?
+		COMMAND_PLAYBACK_STREAM_KILLED :
+		COMMAND_RECORD_STREAM_KILLED;
+
+	pw_log_info("[%s]: %s channel:%u",
+		    client->name, commands[command].name, stream->channel);
+
+	if (client->version < 23)
+		return 0;
+
+	reply = message_alloc(impl, -1, 0);
+	message_put(reply,
+		TAG_U32, command,
+		TAG_U32, -1,
+		TAG_U32, stream->channel,
+		TAG_INVALID);
+
+	return client_queue_message(client, reply);
+}
+
+int stream_send_started(struct stream *stream)
+{
+	struct client *client = stream->client;
+	struct impl *impl = client->impl;
+	struct message *reply;
+
+	pw_log_debug("client %p [%s]: stream %p STARTED channel:%u",
+		     client, client->name, stream, stream->channel);
+
+	reply = message_alloc(impl, -1, 0);
+	message_put(reply,
+		TAG_U32, COMMAND_STARTED,
+		TAG_U32, -1,
+		TAG_U32, stream->channel,
+		TAG_INVALID);
+
+	return client_queue_message(client, reply);
+}
+
+int stream_send_request(struct stream *stream)
+{
+	struct client *client = stream->client;
+	struct impl *impl = client->impl;
+	struct message *msg;
+	uint32_t size;
+
+	size = stream_pop_missing(stream);
+	pw_log_debug("stream %p: REQUEST channel:%d %u", stream, stream->channel, size);
+
+	if (size == 0)
+		return 0;
+
+	msg = message_alloc(impl, -1, 0);
+	message_put(msg,
+		TAG_U32, COMMAND_REQUEST,
+		TAG_U32, -1,
+		TAG_U32, stream->channel,
+		TAG_U32, size,
+		TAG_INVALID);
+
+	return client_queue_message(client, msg);
+}
+
+int stream_update_minreq(struct stream *stream, uint32_t minreq)
+{
+	struct client *client = stream->client;
+	struct impl *impl = client->impl;
+	uint32_t old_tlength = stream->attr.tlength;
+	uint32_t new_tlength = minreq + 2 * stream->attr.minreq;
+	uint64_t lat_usec;
+
+	if (new_tlength <= old_tlength)
+		return 0;
+
+	if (new_tlength > MAXLENGTH)
+		new_tlength = MAXLENGTH;
+
+	stream->attr.tlength = new_tlength;
+	if (stream->attr.tlength > stream->attr.maxlength)
+		stream->attr.maxlength = stream->attr.tlength;
+
+	if (client->version >= 15) {
+		struct message *msg;
+
+		lat_usec = minreq * SPA_USEC_PER_SEC / stream->ss.rate;
+
+		msg = message_alloc(impl, -1, 0);
+		message_put(msg,
+			TAG_U32, COMMAND_PLAYBACK_BUFFER_ATTR_CHANGED,
+			TAG_U32, -1,
+			TAG_U32, stream->channel,
+			TAG_U32, stream->attr.maxlength,
+			TAG_U32, stream->attr.tlength,
+			TAG_U32, stream->attr.prebuf,
+			TAG_U32, stream->attr.minreq,
+			TAG_USEC, lat_usec,
+			TAG_INVALID);
+		return client_queue_message(client, msg);
+	}
+	return 0;
+}
+
+int stream_send_moved(struct stream *stream, uint32_t peer_index, const char *peer_name)
+{
+	struct client *client = stream->client;
+	struct impl *impl = client->impl;
+	struct message *reply;
+	uint32_t command;
+
+	command = stream->direction == PW_DIRECTION_OUTPUT ?
+		COMMAND_PLAYBACK_STREAM_MOVED :
+		COMMAND_RECORD_STREAM_MOVED;
+
+	pw_log_info("client %p [%s]: stream %p %s channel:%u",
+			client, client->name, stream, commands[command].name,
+			stream->channel);
+
+	if (client->version < 12)
+		return 0;
+
+	reply = message_alloc(impl, -1, 0);
+	message_put(reply,
+		TAG_U32, command,
+		TAG_U32, -1,
+		TAG_U32, stream->channel,
+		TAG_U32, peer_index,
+		TAG_STRING, peer_name,
+		TAG_BOOLEAN, false,		/* suspended */
+		TAG_INVALID);
+
+	if (client->version >= 13) {
+		if (command == COMMAND_PLAYBACK_STREAM_MOVED) {
+			message_put(reply,
+				TAG_U32, stream->attr.maxlength,
+				TAG_U32, stream->attr.tlength,
+				TAG_U32, stream->attr.prebuf,
+				TAG_U32, stream->attr.minreq,
+				TAG_USEC, stream->lat_usec,
+				TAG_INVALID);
+		} else {
+			message_put(reply,
+				TAG_U32, stream->attr.maxlength,
+				TAG_U32, stream->attr.fragsize,
+				TAG_USEC, stream->lat_usec,
+				TAG_INVALID);
+		}
+	}
+	return client_queue_message(client, reply);
+}