bluez: add sco-sink and sco-source nodes

6 files changed, 2352 insertions, 98 deletions

diff --git a/spa/include/spa/utils/names.h b/spa/include/spa/utils/names.h
index 8293f72e..a81c541a 100644
--- a/spa/include/spa/utils/names.h
+++ b/spa/include/spa/utils/names.h
@@ -87,8 +87,10 @@ extern "C" {
 /** keys for bluez5 factory names */
 #define SPA_NAME_API_BLUEZ5_MONITOR	"api.bluez5.monitor"		/**< a Monitor interface */
 #define SPA_NAME_API_BLUEZ5_DEVICE	"api.bluez5.device"		/**< a Device interface */
-#define SPA_NAME_API_BLUEZ5_A2DP_SINK	"api.bluez5.a2dp.sink"		/**< a playback Node interface */
-#define SPA_NAME_API_BLUEZ5_A2DP_SOURCE	"api.bluez5.a2dp.source"	/**< a capture Node interface */
+#define SPA_NAME_API_BLUEZ5_A2DP_SINK	"api.bluez5.a2dp.sink"		/**< a playback Node interface for A2DP profiles */
+#define SPA_NAME_API_BLUEZ5_A2DP_SOURCE	"api.bluez5.a2dp.source"	/**< a capture Node interface for A2DP profiles */
+#define SPA_NAME_API_BLUEZ5_SCO_SINK	"api.bluez5.sco.sink"		/**< a playback Node interface for HSP/HFP profiles */
+#define SPA_NAME_API_BLUEZ5_SCO_SOURCE	"api.bluez5.sco.source"		/**< a capture Node interface for HSP/HFP profiles */
 
 /** keys for v4l2 factory names */
 #define SPA_NAME_API_V4L2_MONITOR	"api.v4l2.monitor"		/**< a v4l2 Monitor interface */
diff --git a/spa/plugins/bluez5/bluez5-device.c b/spa/plugins/bluez5/bluez5-device.c
index 768f7ee6..48dbebb5 100644
--- a/spa/plugins/bluez5/bluez5-device.c
+++ b/spa/plugins/bluez5/bluez5-device.c
@@ -67,120 +67,58 @@ struct impl {
 	struct props props;
 
 	struct spa_bt_device *bt_dev;
+
+	uint32_t next_id;
 };
 
-static int emit_source_node(struct impl *this)
+static void emit_node (struct impl *this, struct spa_bt_transport *t, const char *factory_name)
 {
-	struct spa_dict_item items[1];
-	struct spa_bt_transport *t;
-	struct spa_bt_device *device = this->bt_dev;
-	enum spa_bt_profile profile = SPA_BT_PROFILE_NULL;
-
-	if (device->connected_profiles & SPA_BT_PROFILE_A2DP_SOURCE) {
-		spa_log_info(this->log, "device %p: A2DP (source) profile found",
-				device);
-		profile = SPA_BT_PROFILE_A2DP_SOURCE;
-	} else if (device->connected_profiles & SPA_BT_PROFILE_HSP_HS) {
-		spa_log_info(this->log, "device %p: HSP (source) profile found (Not implemented yet)",
-				device);
-		profile = SPA_BT_PROFILE_HSP_HS;
-		return -ENODEV;
-	} else if (device->connected_profiles & SPA_BT_PROFILE_HFP_HF) {
-		spa_log_info(this->log, "device %p: HFP (source) profile found (Not implemented yet)",
-				device);
-		profile = SPA_BT_PROFILE_HFP_HF;
-		return -ENODEV;
-	}
-
-	/* Return if no profiles are connected */
-	if (profile == SPA_BT_PROFILE_NULL)
-		return -ENODEV;
-
-	spa_list_for_each(t, &device->transport_list, device_link) {
-		if (t->profile == profile) {
-			struct spa_device_object_info info;
-			char transport[32];
-
-			snprintf(transport, sizeof(transport), "pointer:%p", t);
-			items[0] = SPA_DICT_ITEM_INIT(SPA_KEY_API_BLUEZ5_TRANSPORT, transport);
+        struct spa_device_object_info info;
+        struct spa_dict_item items[1];
+        char transport[32];
 
-			spa_bt_transport_acquire(t, true);
+        snprintf(transport, sizeof(transport), "pointer:%p", t);
+        items[0] = SPA_DICT_ITEM_INIT(SPA_KEY_API_BLUEZ5_TRANSPORT, transport);
 
-			info = SPA_DEVICE_OBJECT_INFO_INIT();
-			info.type = SPA_TYPE_INTERFACE_Node;
-			info.factory_name = SPA_NAME_API_BLUEZ5_A2DP_SOURCE;
-			info.change_mask = SPA_DEVICE_OBJECT_CHANGE_MASK_PROPS;
-			info.props = &SPA_DICT_INIT_ARRAY(items);
+        info = SPA_DEVICE_OBJECT_INFO_INIT();
+        info.type = SPA_TYPE_INTERFACE_Node;
+        info.factory_name = factory_name;
+        info.change_mask = SPA_DEVICE_OBJECT_CHANGE_MASK_PROPS;
+        info.props = &SPA_DICT_INIT_ARRAY(items);
 
-			spa_device_emit_object_info(&this->hooks, 0, &info);
-			break;
-		}
-	}
-
-	spa_log_info (this->log, "device %p: bluez5 source nodes emitted", device);
-	return 0;
+        spa_device_emit_object_info(&this->hooks, this->next_id++, &info);
 }
 
-static int emit_sink_node(struct impl *this)
+static int emit_nodes(struct impl *this)
 {
-	struct spa_dict_item items[1];
-	struct spa_bt_transport *t;
 	struct spa_bt_device *device = this->bt_dev;
-	enum spa_bt_profile profile = SPA_BT_PROFILE_NULL;
-
-	if (device->connected_profiles & SPA_BT_PROFILE_A2DP_SINK) {
-		spa_log_info(this->log, "A2DP (sink) profile found");
-		profile = SPA_BT_PROFILE_A2DP_SINK;
-	} else if (device->connected_profiles & SPA_BT_PROFILE_HSP_AG) {
-		spa_log_info(this->log, "HSP (sink) profile found (Not implemented yet)");
-		profile = SPA_BT_PROFILE_HSP_AG;
-		return -ENODEV;
-	} else if (device->connected_profiles & SPA_BT_PROFILE_HFP_AG) {
-		spa_log_info(this->log, "HFP (sink) profile found (Not implemented yet)");
-		profile = SPA_BT_PROFILE_HFP_AG;
-		return -ENODEV;
-	}
-
-	/* Return if no profiles are connected */
-	if (profile == SPA_BT_PROFILE_NULL)
-		return -ENODEV;
+	struct spa_bt_transport *t;
 
 	spa_list_for_each(t, &device->transport_list, device_link) {
-		if (t->profile == profile) {
-			struct spa_device_object_info info;
-			char transport[32];
-
-			snprintf(transport, sizeof(transport), "pointer:%p", t);
-			items[0] = SPA_DICT_ITEM_INIT(SPA_KEY_API_BLUEZ5_TRANSPORT, transport);
-
-			info = SPA_DEVICE_OBJECT_INFO_INIT();
-			info.type = SPA_TYPE_INTERFACE_Node;
-			info.factory_name = SPA_NAME_API_BLUEZ5_A2DP_SINK;
-			info.change_mask = SPA_DEVICE_OBJECT_CHANGE_MASK_PROPS;
-			info.props = &SPA_DICT_INIT_ARRAY(items);
-
-			spa_device_emit_object_info(&this->hooks, 0, &info);
-			break;
+		if (t->profile & device->connected_profiles) {
+			switch (t->profile) {
+			case SPA_BT_PROFILE_A2DP_SOURCE:
+				emit_node (this, t, SPA_NAME_API_BLUEZ5_A2DP_SOURCE);
+				break;
+			case SPA_BT_PROFILE_A2DP_SINK:
+				emit_node (this, t, SPA_NAME_API_BLUEZ5_A2DP_SINK);
+				break;
+			case SPA_BT_PROFILE_HSP_HS:
+			case SPA_BT_PROFILE_HSP_AG:
+			case SPA_BT_PROFILE_HFP_HF:
+			case SPA_BT_PROFILE_HFP_AG:
+				emit_node (this, t, SPA_NAME_API_BLUEZ5_SCO_SOURCE);
+				emit_node (this, t, SPA_NAME_API_BLUEZ5_SCO_SINK);
+				break;
+			default:
+				return -EINVAL;
+			}
 		}
 	}
 
-	spa_log_info(this->log, "bluez5 sink nodes emitted");
 	return 0;
 }
 
-static int emit_nodes(struct impl *this)
-{
-	int sink, src;
-
-	sink = emit_sink_node(this);
-	src = emit_source_node(this);
-
-	if (sink == -ENODEV && src == -ENODEV)
-		spa_log_warn(this->log, "no profile available");
-
-	return SPA_MAX(sink, src);
-}
-
 static const struct spa_dict_item info_items[] = {
 	{ SPA_KEY_DEVICE_API, "bluez5" },
 	{ SPA_KEY_MEDIA_CLASS, "Audio/Device" },
@@ -316,6 +254,8 @@ impl_init(const struct spa_handle_factory *factory,
 
 	reset_props(&this->props);
 
+	this->next_id = 0;
+
 	return 0;
 }
 
diff --git a/spa/plugins/bluez5/meson.build b/spa/plugins/bluez5/meson.build
index 5fb285ec..ddcc74fa 100644
--- a/spa/plugins/bluez5/meson.build
+++ b/spa/plugins/bluez5/meson.build
@@ -3,6 +3,8 @@ bluez5_sources = ['plugin.c',
 		  'a2dp-codecs.c',
 		  'a2dp-sink.c',
 		  'a2dp-source.c',
+		  'sco-sink.c',
+		  'sco-source.c',
 		  'bluez5-device.c',
                   'bluez5-monitor.c']
 
diff --git a/spa/plugins/bluez5/plugin.c b/spa/plugins/bluez5/plugin.c
index 392416f2..2b70a578 100644
--- a/spa/plugins/bluez5/plugin.c
+++ b/spa/plugins/bluez5/plugin.c
@@ -31,6 +31,8 @@ extern const struct spa_handle_factory spa_bluez5_monitor_factory;
 extern const struct spa_handle_factory spa_bluez5_device_factory;
 extern const struct spa_handle_factory spa_a2dp_sink_factory;
 extern const struct spa_handle_factory spa_a2dp_source_factory;
+extern const struct spa_handle_factory spa_sco_sink_factory;
+extern const struct spa_handle_factory spa_sco_source_factory;
 
 SPA_EXPORT
 int spa_handle_factory_enum(const struct spa_handle_factory **factory, uint32_t *index)
@@ -51,6 +53,12 @@ int spa_handle_factory_enum(const struct spa_handle_factory **factory, uint32_t
 	case 3:
 		*factory = &spa_a2dp_source_factory;
 		break;
+	case 4:
+		*factory = &spa_sco_sink_factory;
+		break;
+	case 5:
+		*factory = &spa_sco_source_factory;
+		break;
 	default:
 		return 0;
 	}
diff --git a/spa/plugins/bluez5/sco-sink.c b/spa/plugins/bluez5/sco-sink.c
new file mode 100644
index 00000000..b91ec5be
--- /dev/null
+++ b/spa/plugins/bluez5/sco-sink.c
@@ -0,0 +1,1186 @@
+/* Spa SCO Sink
+ *
+ * Copyright © 2019 Collabora Ltd.
+ *
+ * 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 <unistd.h>
+#include <stddef.h>
+#include <stdio.h>
+#include <arpa/inet.h>
+#include <sys/ioctl.h>
+
+#include <spa/support/loop.h>
+#include <spa/support/log.h>
+#include <spa/support/system.h>
+#include <spa/utils/list.h>
+#include <spa/utils/keys.h>
+#include <spa/utils/names.h>
+#include <spa/monitor/device.h>
+
+#include <spa/node/node.h>
+#include <spa/node/utils.h>
+#include <spa/node/io.h>
+#include <spa/param/param.h>
+#include <spa/param/audio/format.h>
+#include <spa/param/audio/format-utils.h>
+#include <spa/pod/filter.h>
+
+#include "defs.h"
+
+struct props {
+	uint32_t min_latency;
+	uint32_t max_latency;
+};
+
+#define MAX_BUFFERS 32
+
+struct buffer {
+	uint32_t id;
+	unsigned int outstanding:1;
+	struct spa_buffer *buf;
+	struct spa_meta_header *h;
+	struct spa_list link;
+};
+
+struct port {
+	struct spa_audio_info current_format;
+	int frame_size;
+	unsigned int have_format:1;
+
+	uint64_t info_all;
+	struct spa_port_info info;
+	struct spa_io_buffers *io;
+	struct spa_param_info params[8];
+
+	struct buffer buffers[MAX_BUFFERS];
+	uint32_t n_buffers;
+
+	struct spa_list free;
+	struct spa_list ready;
+
+	unsigned int need_data:1;
+};
+
+struct impl {
+	struct spa_handle handle;
+	struct spa_node node;
+
+	/* Support */
+	struct spa_log *log;
+	struct spa_loop *data_loop;
+	struct spa_system *data_system;
+
+	/* Hooks and callbacks */
+	struct spa_hook_list hooks;
+	struct spa_callbacks callbacks;
+
+	/* Info */
+	uint64_t info_all;
+	struct spa_node_info info;
+	struct spa_param_info params[8];
+	struct props props;
+
+	/* Transport */
+	struct spa_bt_transport *transport;
+	struct spa_hook transport_listener;
+	int sock_fd;
+
+	/* Port */
+	struct port port;
+
+	/* Flags */
+	unsigned int started:1;
+	unsigned int slaved:1;
+
+	/* Sources */
+	struct spa_source source;
+	struct spa_source flush_source;
+
+	/* Timer */
+	int timerfd;
+	struct timespec now;
+	struct spa_io_clock *clock;
+	struct spa_io_position *position;
+	int threshold;
+
+	/* Times */
+	uint64_t start_time;
+
+	/* Counts */
+	uint64_t sample_count;
+};
+
+#define NAME "sco-sink"
+
+#define CHECK_PORT(this,d,p)    ((d) == SPA_DIRECTION_INPUT && (p) == 0)
+
+static const uint32_t default_min_latency = 128;
+static const uint32_t default_max_latency = 1024;
+
+static void reset_props(struct props *props)
+{
+	props->min_latency = default_min_latency;
+	props->max_latency = default_max_latency;
+}
+
+static int impl_node_enum_params(void *object, int seq,
+			uint32_t id, uint32_t start, uint32_t num,
+			const struct spa_pod *filter)
+{
+	struct impl *this = object;
+	struct spa_pod *param;
+	struct spa_pod_builder b = { 0 };
+	uint8_t buffer[1024];
+	struct spa_result_node_params result;
+	uint32_t count = 0;
+
+	spa_return_val_if_fail(this != NULL, -EINVAL);
+	spa_return_val_if_fail(num != 0, -EINVAL);
+
+	result.id = id;
+	result.next = start;
+      next:
+	result.index = result.next++;
+
+	spa_pod_builder_init(&b, buffer, sizeof(buffer));
+
+	switch (id) {
+	case SPA_PARAM_PropInfo:
+	{
+		struct props *p = &this->props;
+
+		switch (result.index) {
+		case 0:
+			param = spa_pod_builder_add_object(&b,
+				SPA_TYPE_OBJECT_PropInfo, id,
+				SPA_PROP_INFO_id,   SPA_POD_Id(SPA_PROP_minLatency),
+				SPA_PROP_INFO_name, SPA_POD_String("The minimum latency"),
+				SPA_PROP_INFO_type, SPA_POD_CHOICE_RANGE_Int(p->min_latency, 1, INT32_MAX));
+			break;
+		case 1:
+			param = spa_pod_builder_add_object(&b,
+				SPA_TYPE_OBJECT_PropInfo, id,
+				SPA_PROP_INFO_id,   SPA_POD_Id(SPA_PROP_maxLatency),
+				SPA_PROP_INFO_name, SPA_POD_String("The maximum latency"),
+				SPA_PROP_INFO_type, SPA_POD_CHOICE_RANGE_Int(p->max_latency, 1, INT32_MAX));
+			break;
+		default:
+			return 0;
+		}
+		break;
+	}
+	case SPA_PARAM_Props:
+	{
+		struct props *p = &this->props;
+
+		switch (result.index) {
+		case 0:
+			param = spa_pod_builder_add_object(&b,
+				SPA_TYPE_OBJECT_Props, id,
+				SPA_PROP_minLatency, SPA_POD_Int(p->min_latency),
+				SPA_PROP_maxLatency, SPA_POD_Int(p->max_latency));
+			break;
+		default:
+			return 0;
+		}
+		break;
+	}
+	default:
+		return -ENOENT;
+	}
+
+	if (spa_pod_filter(&b, &result.param, param, filter) < 0)
+		goto next;
+
+	spa_node_emit_result(&this->hooks, seq, 0, SPA_RESULT_TYPE_NODE_PARAMS, &result);
+
+	if (++count != num)
+		goto next;
+
+	return 0;
+}
+
+static void set_timeout(struct impl *this, time_t sec, long nsec)
+{
+	struct itimerspec ts;
+
+	ts.it_value.tv_sec = sec;
+	ts.it_value.tv_nsec = nsec;
+	ts.it_interval.tv_sec = 0;
+	ts.it_interval.tv_nsec = 0;
+
+	timerfd_settime(this->timerfd, TFD_TIMER_ABSTIME, &ts, NULL);
+	this->source.mask = SPA_IO_IN;
+	spa_loop_update_source(this->data_loop, &this->source);
+}
+
+static void reset_timeout(struct impl *this)
+{
+	set_timeout(this, 0, this->slaved ? 0 : 1);
+}
+
+
+static void set_next_timeout(struct impl *this, uint64_t now_time)
+{
+	struct port *port = &this->port;
+
+	/* Set the next timeout if not slaved, otherwise reset values */
+	if (!this->slaved) {
+		/* Get the elapsed time */
+		const uint64_t elapsed_time = now_time - this->start_time;
+
+		/* Get the elapsed samples */
+		const uint64_t elapsed_samples = elapsed_time * port->current_format.info.raw.rate / SPA_NSEC_PER_SEC;
+
+		/* Get the queued samples (processed - elapsed) */
+		const uint64_t queued_samples = this->sample_count - elapsed_samples;
+
+		/* Get the queued time */
+		const uint64_t queued_time = (queued_samples * SPA_NSEC_PER_SEC) / port->current_format.info.raw.rate;
+
+		/* Get the next time */
+		const uint64_t next_time = now_time + queued_time;
+
+		/* Set the next timeout */
+		set_timeout (this, next_time / SPA_NSEC_PER_SEC, next_time % SPA_NSEC_PER_SEC);
+	} else {
+		this->start_time = now_time;
+		this->sample_count = 0;
+	}
+}
+
+static int do_reslave(struct spa_loop *loop,
+			bool async,
+			uint32_t seq,
+			const void *data,
+			size_t size,
+			void *user_data)
+{
+	struct impl *this = user_data;
+	reset_timeout(this);
+	return 0;
+}
+
+static inline bool is_slaved(struct impl *this)
+{
+	return this->position && this->clock && this->position->clock.id != this->clock->id;
+}
+
+static int impl_node_set_io(void *object, uint32_t id, void *data, size_t size)
+{
+	struct impl *this = object;
+	bool slaved;
+
+	spa_return_val_if_fail(object != NULL, -EINVAL);
+
+	switch (id) {
+	case SPA_IO_Clock:
+		this->clock = data;
+		break;
+	case SPA_IO_Position:
+		this->position = data;
+		break;
+	default:
+		return -ENOENT;
+	}
+
+	slaved = is_slaved(this);
+	if (this->started && slaved != this->slaved) {
+		spa_log_debug(this->log, "sco-sink %p: reslave %d->%d", this, this->slaved, slaved);
+		this->slaved = slaved;
+		spa_loop_invoke(this->data_loop, do_reslave, 0, NULL, 0, true, this);
+	}
+	return 0;
+}
+
+static int impl_node_set_param(void *object, uint32_t id, uint32_t flags,
+			       const struct spa_pod *param)
+{
+	struct impl *this = object;
+
+	spa_return_val_if_fail(this != NULL, -EINVAL);
+
+	switch (id) {
+	case SPA_PARAM_Props:
+	{
+		struct props *p = &this->props;
+
+		if (param == NULL) {
+			reset_props(p);
+			return 0;
+		}
+		spa_pod_parse_object(param,
+			SPA_TYPE_OBJECT_Props, NULL,
+			SPA_PROP_minLatency, SPA_POD_OPT_Int(&p->min_latency),
+			SPA_PROP_maxLatency, SPA_POD_OPT_Int(&p->max_latency));
+		break;
+	}
+	default:
+		return -ENOENT;
+	}
+
+	return 0;
+}
+
+static bool write_data(struct impl *this, uint8_t *data, uint32_t size, uint32_t *total_written)
+{
+	uint32_t local_total_written = 0;
+	const uint32_t mtu_size = this->transport->write_mtu;
+
+	/* TODO: For now we assume the size is always a mutliple of mtu_size */
+	while (local_total_written < (size - mtu_size)) {
+		const int bytes_written = write(this->sock_fd, data, mtu_size);
+		if (bytes_written < 0) {
+			spa_log_warn(this->log, "error writting data: %s", strerror(errno));
+			return false;
+		}
+
+		data += bytes_written;
+		local_total_written += bytes_written;
+	}
+
+	if (total_written)
+		*total_written = local_total_written;
+	return true;
+}
+
+static int render_buffers(struct impl *this, uint64_t now_time)
+{
+	struct port *port = &this->port;
+
+	/* Render the buffer */
+	while (!spa_list_is_empty(&port->ready)) {
+		uint8_t *src;
+		struct buffer *b;
+		struct spa_data *d;
+		uint32_t offset, size;
+		uint32_t total_written = 0;
+
+		/* Get the buffer and datas */
+		b = spa_list_first(&port->ready, struct buffer, link);
+		d = b->buf->datas;
+
+		/* Get the data, offset and size */
+		src = d[0].data;
+		offset = d[0].chunk->offset;
+		size = d[0].chunk->size;
+
+		/* Write data */
+		write_data(this, src + offset, size, &total_written);
+
+		/* Update the cample count */
+		this->sample_count += total_written / port->frame_size;
+
+		/* Remove the buffer and mark it as reusable */
+		spa_list_remove(&b->link);
+		b->outstanding = true;
+		spa_node_call_reuse_buffer(&this->callbacks, 0, b->id);
+	}
+
+	/* Set next timeout */
+	set_next_timeout(this, now_time);
+
+	return 0;
+}
+
+static void sco_on_flush(struct spa_source *source)
+{
+	struct impl *this = source->data;
+	uint64_t now_time;
+
+	spa_log_trace(this->log, NAME" %p: flushing", this);
+
+	if ((source->rmask & SPA_IO_OUT) == 0) {
+		spa_log_warn(this->log, "error %d", source->rmask);
+		if (this->flush_source.loop)
+			spa_loop_remove_source(this->data_loop, &this->flush_source);
+		this->source.mask = 0;
+		spa_loop_update_source(this->data_loop, &this->source);
+		return;
+	}
+
+	/* Get the current time */
+	spa_system_clock_gettime(this->data_system, CLOCK_MONOTONIC, &this->now);
+	now_time = SPA_TIMESPEC_TO_NSEC(&this->now);
+
+	/* Render buffers */
+	render_buffers(this, now_time);
+}
+
+static void sco_on_timeout(struct spa_source *source)
+{
+	struct impl *this = source->data;
+	struct port *port = &this->port;
+	uint64_t exp, now_time;
+	struct spa_io_buffers *io = port->io;
+
+	/* Read the timerfd */
+	if (this->started && read(this->timerfd, &exp, sizeof(uint64_t)) != sizeof(uint64_t))
+		spa_log_warn(this->log, "error reading timerfd: %s", strerror(errno));
+
+	/* Get the current time */
+	spa_system_clock_gettime(this->data_system, CLOCK_MONOTONIC, &this->now);
+	now_time = SPA_TIMESPEC_TO_NSEC(&this->now);
+
+	/* Set the start time to the current time */
+	if (this->start_time == 0)
+		this->start_time = now_time;
+
+	/* Notify we need a new buffer if we have processed all of them */
+	if (spa_list_is_empty(&port->ready) || port->need_data) {
+		io->status = SPA_STATUS_NEED_BUFFER;
+		spa_node_call_ready(&this->callbacks, SPA_STATUS_NEED_BUFFER);
+	}
+
+	/* Render the buffers */
+	render_buffers(this, now_time);
+}
+
+static int do_start(struct impl *this)
+{
+	int val;
+	bool do_accept;
+
+	/* Dont do anything if the node has already started */
+	if (this->started)
+		return 0;
+
+	/* Make sure the transport is valid */
+	spa_return_val_if_fail (this->transport != NULL, -EIO);
+
+	/* Set the slaved flag */
+	this->slaved = is_slaved(this);
+
+	/* Do accept if Gateway; otherwise do connect for Head Unit */
+	do_accept = this->transport->profile & SPA_BT_PROFILE_HEADSET_AUDIO_GATEWAY;
+
+	/* acquire the socked fd (false -> connect | true -> accept) */
+	this->sock_fd = spa_bt_transport_acquire(this->transport, do_accept);
+	if (this->sock_fd < 0)
+		return -1;
+
+	/* Set the write MTU */
+	val = this->transport->write_mtu;
+	if (setsockopt(this->sock_fd, SOL_SOCKET, SO_SNDBUF, &val, sizeof(val)) < 0)
+		spa_log_warn(this->log, "sco-sink %p: SO_SNDBUF %m", this);
+
+	/* Set the read MTU */
+	val = this->transport->read_mtu;
+	if (setsockopt(this->sock_fd, SOL_SOCKET, SO_RCVBUF, &val, sizeof(val)) < 0)
+		spa_log_warn(this->log, "sco-sink %p: SO_RCVBUF %m", this);
+
+	/* Set the priority */
+	val = 6;
+	if (setsockopt(this->sock_fd, SOL_SOCKET, SO_PRIORITY, &val, sizeof(val)) < 0)
+		spa_log_warn(this->log, "SO_PRIORITY failed: %m");
+
+	/* Add the timeout callback */
+	this->source.data = this;
+	this->source.fd = this->timerfd;
+	this->source.func = sco_on_timeout;
+	this->source.mask = SPA_IO_IN;
+	this->source.rmask = 0;
+	spa_loop_add_source(this->data_loop, &this->source);
+
+	/* Add the flush callback */
+	this->flush_source.data = this;
+	this->flush_source.fd = this->sock_fd;
+	this->flush_source.func = sco_on_flush;
+	this->flush_source.mask = 0;
+	this->flush_source.rmask = 0;
+	spa_loop_add_source(this->data_loop, &this->flush_source);
+
+	/* Reset timeout to start processing */
+	reset_timeout(this);
+
+	/* Set the started flag */
+	this->started = true;
+
+	return 0;
+}
+
+static int do_remove_source(struct spa_loop *loop,
+			    bool async,
+			    uint32_t seq,
+			    const void *data,
+			    size_t size,
+			    void *user_data)
+{
+	struct impl *this = user_data;
+	struct itimerspec ts;
+
+	if (this->source.loop)
+		spa_loop_remove_source(this->data_loop, &this->source);
+	ts.it_value.tv_sec = 0;
+	ts.it_value.tv_nsec = 0;
+	ts.it_interval.tv_sec = 0;
+	ts.it_interval.tv_nsec = 0;
+	timerfd_settime(this->timerfd, 0, &ts, NULL);
+	if (this->flush_source.loop)
+		spa_loop_remove_source(this->data_loop, &this->flush_source);
+
+	return 0;
+}
+
+static int do_stop(struct impl *this)
+{
+	int res = 0;
+
+	if (!this->started)
+		return 0;
+
+	spa_log_trace(this->log, "sco-sink %p: stop", this);
+
+	spa_loop_invoke(this->data_loop, do_remove_source, 0, NULL, 0, true, this);
+
+	this->started = false;
+
+	if (this->transport) {
+		/* Release the transport */
+		res = spa_bt_transport_release(this->transport);
+
+		/* Shutdown and close the socket */
+		shutdown(this->sock_fd, SHUT_RDWR);
+		close(this->sock_fd);
+		this->sock_fd = -1;
+	}
+
+	return res;
+}
+
+static int impl_node_send_command(void *object, const struct spa_command *command)
+{
+	struct impl *this = object;
+	struct port *port;
+	int res;
+
+	spa_return_val_if_fail(this != NULL, -EINVAL);
+	spa_return_val_if_fail(command != NULL, -EINVAL);
+
+	port = &this->port;
+
+	switch (SPA_NODE_COMMAND_ID(command)) {
+	case SPA_NODE_COMMAND_Start:
+		if (!port->have_format)
+			return -EIO;
+		if (port->n_buffers == 0)
+			return -EIO;
+		if ((res = do_start(this)) < 0)
+			return res;
+		break;
+	case SPA_NODE_COMMAND_Pause:
+		if ((res = do_stop(this)) < 0)
+			return res;
+		break;
+	default:
+		return -ENOTSUP;
+	}
+	return 0;
+}
+
+static const struct spa_dict_item node_info_items[] = {
+	{ SPA_KEY_DEVICE_API, "bluez5" },
+	{ SPA_KEY_MEDIA_CLASS, "Audio/Sink" },
+	{ SPA_KEY_NODE_DRIVER, "true" },
+};
+
+static void emit_node_info(struct impl *this, bool full)
+{
+	if (full)
+		this->info.change_mask = this->info_all;
+	if (this->info.change_mask) {
+		this->info.props = &SPA_DICT_INIT_ARRAY(node_info_items);
+		spa_node_emit_info(&this->hooks, &this->info);
+		this->info.change_mask = 0;
+	}
+}
+
+static void emit_port_info(struct impl *this, struct port *port, bool full)
+{
+	if (full)
+		port->info.change_mask = port->info_all;
+	if (port->info.change_mask) {
+		spa_node_emit_port_info(&this->hooks,
+				SPA_DIRECTION_INPUT, 0, &port->info);
+		port->info.change_mask = 0;
+	}
+}
+
+static int
+impl_node_add_listener(void *object,
+		struct spa_hook *listener,
+		const struct spa_node_events *events,
+		void *data)
+{
+	struct impl *this = object;
+	struct spa_hook_list save;
+
+	spa_return_val_if_fail(this != NULL, -EINVAL);
+
+	spa_hook_list_isolate(&this->hooks, &save, listener, events, data);
+
+	emit_node_info(this, true);
+	emit_port_info(this, &this->port, true);
+
+	spa_hook_list_join(&this->hooks, &save);
+
+	return 0;
+}
+
+static int
+impl_node_set_callbacks(void *object,
+			const struct spa_node_callbacks *callbacks,
+			void *data)
+{
+	struct impl *this = object;
+
+	spa_return_val_if_fail(this != NULL, -EINVAL);
+
+	this->callbacks = SPA_CALLBACKS_INIT(callbacks, data);
+
+	return 0;
+}
+
+static int impl_node_sync(void *object, int seq)
+{
+	struct impl *this = object;
+
+	spa_return_val_if_fail(this != NULL, -EINVAL);
+
+	spa_node_emit_result(&this->hooks, seq, 0, 0, NULL);
+
+	return 0;
+}
+
+static int impl_node_add_port(void *object, enum spa_direction direction, uint32_t port_id,
+		const struct spa_dict *props)
+{
+	return -ENOTSUP;
+}
+
+static int impl_node_remove_port(void *object, enum spa_direction direction, uint32_t port_id)
+{
+	return -ENOTSUP;
+}
+
+static int
+impl_node_port_enum_params(void *object, int seq,
+			enum spa_direction direction, uint32_t port_id,
+			uint32_t id, uint32_t start, uint32_t num,
+			const struct spa_pod *filter)
+{
+
+	struct impl *this = object;
+	struct port *port;
+	struct spa_pod *param;
+	struct spa_pod_builder b = { 0 };
+	uint8_t buffer[1024];
+	struct spa_result_node_params result;
+	uint32_t count = 0;
+
+	spa_return_val_if_fail(this != NULL, -EINVAL);
+	spa_return_val_if_fail(num != 0, -EINVAL);
+
+	spa_return_val_if_fail(CHECK_PORT(this, direction, port_id), -EINVAL);
+	port = &this->port;
+
+	result.id = id;
+	result.next = start;
+      next:
+	result.index = result.next++;
+
+	spa_pod_builder_init(&b, buffer, sizeof(buffer));
+
+	switch (id) {
+	case SPA_PARAM_EnumFormat:
+		if (result.index > 0)
+			return 0;
+
+		/* set the info structure */
+		struct spa_audio_info_raw info = { 0, };
+		info.format = SPA_AUDIO_FORMAT_S16;
+		info.channels = 1;
+		info.position[0] = SPA_AUDIO_CHANNEL_MONO;
+
+		/* TODO: For now we only handle HSP profiles which has always CVSD format,
+		 * but we eventually need to support HFP that can have both CVSD and MSBC formats */
+
+		 /* CVSD format has a rate of 8kHz
+		  * MSBC format has a rate of 16kHz */
+		info.rate = 8000;
+
+		/* build the param */
+		param = spa_format_audio_raw_build(&b, id, &info);
+
+		break;
+
+	case SPA_PARAM_Format:
+		if (!port->have_format)
+			return -EIO;
+		if (result.index > 0)
+			return 0;
+
+		param = spa_format_audio_raw_build(&b, id, &port->current_format.info.raw);
+		break;
+
+	case SPA_PARAM_Buffers:
+		if (!port->have_format)
+			return -EIO;
+		if (result.index > 0)
+			return 0;
+
+		param = spa_pod_builder_add_object(&b,
+			SPA_TYPE_OBJECT_ParamBuffers, id,
+			SPA_PARAM_BUFFERS_buffers, SPA_POD_CHOICE_RANGE_Int(2, 2, MAX_BUFFERS),
+			SPA_PARAM_BUFFERS_blocks,  SPA_POD_Int(1),
+			SPA_PARAM_BUFFERS_size,    SPA_POD_CHOICE_RANGE_Int(
+							this->props.min_latency * port->frame_size,
+							this->props.min_latency * port->frame_size,
+							INT32_MAX),
+			SPA_PARAM_BUFFERS_stride,  SPA_POD_Int(port->frame_size),
+			SPA_PARAM_BUFFERS_align,   SPA_POD_Int(16));
+		break;
+
+	case SPA_PARAM_Meta:
+		switch (result.index) {
+		case 0:
+			param = spa_pod_builder_add_object(&b,
+				SPA_TYPE_OBJECT_ParamMeta, id,
+				SPA_PARAM_META_type, SPA_POD_Id(SPA_META_Header),
+				SPA_PARAM_META_size, SPA_POD_Int(sizeof(struct spa_meta_header)));
+			break;
+		default:
+			return 0;
+		}
+		break;
+
+	default:
+		return -ENOENT;
+	}
+
+	if (spa_pod_filter(&b, &result.param, param, filter) < 0)
+		goto next;
+
+	spa_node_emit_result(&this->hooks, seq, 0, SPA_RESULT_TYPE_NODE_PARAMS, &result);
+
+	if (++count != num)
+		goto next;
+
+	return 0;
+}
+
+static int clear_buffers(struct impl *this, struct port *port)
+{
+	do_stop(this);
+	if (port->n_buffers > 0) {
+		spa_list_init(&port->ready);
+		port->n_buffers = 0;
+	}
+	return 0;
+}
+
+static int port_set_format(struct impl *this, struct port *port,
+			   uint32_t flags,
+			   const struct spa_pod *format)
+{
+	int err;
+
+	if (format == NULL) {
+		spa_log_info(this->log, "clear format");
+		clear_buffers(this, port);
+		port->have_format = false;
+	} else {
+		struct spa_audio_info info = { 0 };
+
+		if ((err = spa_format_parse(format, &info.media_type, &info.media_subtype)) < 0)
+			return err;
+
+		if (info.media_type != SPA_MEDIA_TYPE_audio ||
+		    info.media_subtype != SPA_MEDIA_SUBTYPE_raw)
+			return -EINVAL;
+
+		if (spa_format_audio_raw_parse(format, &info.info.raw) < 0)
+			return -EINVAL;
+
+		port->frame_size = info.info.raw.channels * 2;
+		port->current_format = info;
+		port->have_format = true;
+		this->threshold = this->props.min_latency;
+	}
+
+	port->info.change_mask |= SPA_PORT_CHANGE_MASK_PARAMS;
+	if (port->have_format) {
+		port->info.change_mask |= SPA_PORT_CHANGE_MASK_FLAGS;
+		port->info.flags = SPA_PORT_FLAG_LIVE;
+		port->info.change_mask |= SPA_PORT_CHANGE_MASK_RATE;
+		port->info.rate = SPA_FRACTION(1, port->current_format.info.raw.rate);
+		port->params[3] = SPA_PARAM_INFO(SPA_PARAM_Format, SPA_PARAM_INFO_READWRITE);
+		port->params[4] = SPA_PARAM_INFO(SPA_PARAM_Buffers, SPA_PARAM_INFO_READ);
+	} else {
+		port->params[3] = SPA_PARAM_INFO(SPA_PARAM_Format, SPA_PARAM_INFO_WRITE);
+		port->params[4] = SPA_PARAM_INFO(SPA_PARAM_Buffers, 0);
+	}
+	emit_port_info(this, port, false);
+
+	return 0;
+}
+
+static int
+impl_node_port_set_param(void *object,
+			 enum spa_direction direction, uint32_t port_id,
+			 uint32_t id, uint32_t flags,
+			 const struct spa_pod *param)
+{
+	struct impl *this = object;
+	struct port *port;
+	int res;
+
+	spa_return_val_if_fail(this != NULL, -EINVAL);
+	spa_return_val_if_fail(CHECK_PORT(node, direction, port_id), -EINVAL);
+	port = &this->port;
+
+	switch (id) {
+	case SPA_PARAM_Format:
+		res = port_set_format(this, port, flags, param);
+		break;
+	default:
+		res = -ENOENT;
+		break;
+	}
+	return res;
+}
+
+static int
+impl_node_port_use_buffers(void *object,
+			   enum spa_direction direction, uint32_t port_id,
+			   uint32_t flags,
+			   struct spa_buffer **buffers, uint32_t n_buffers)
+{
+	struct impl *this = object;
+	struct port *port;
+	uint32_t i;
+
+	spa_return_val_if_fail(this != NULL, -EINVAL);
+	spa_return_val_if_fail(CHECK_PORT(this, direction, port_id), -EINVAL);
+	port = &this->port;
+
+	spa_log_info(this->log, "use buffers %d", n_buffers);
+
+	if (!port->have_format)
+		return -EIO;
+
+	clear_buffers(this, port);
+
+	for (i = 0; i < n_buffers; i++) {
+		struct buffer *b = &port->buffers[i];
+		uint32_t type;
+
+		b->buf = buffers[i];
+		b->id = i;
+		b->outstanding = true;
+
+		b->h = spa_buffer_find_meta_data(buffers[i], SPA_META_Header, sizeof(*b->h));
+
+		type = buffers[i]->datas[0].type;
+		if ((type == SPA_DATA_MemFd ||
+		     type == SPA_DATA_DmaBuf ||
+		     type == SPA_DATA_MemPtr) && buffers[i]->datas[0].data == NULL) {
+			spa_log_error(this->log, NAME " %p: need mapped memory", this);
+			return -EINVAL;
+		}
+		this->threshold = buffers[i]->datas[0].maxsize / port->frame_size;
+	}
+	port->n_buffers = n_buffers;
+
+	return 0;
+}
+
+static int
+impl_node_port_set_io(void *object,
+		      enum spa_direction direction,
+		      uint32_t port_id,
+		      uint32_t id,
+		      void *data, size_t size)
+{
+	struct impl *this = object;
+	struct port *port;
+
+	spa_return_val_if_fail(this != NULL, -EINVAL);
+
+	spa_return_val_if_fail(CHECK_PORT(this, direction, port_id), -EINVAL);
+	port = &this->port;
+
+	switch (id) {
+	case SPA_IO_Buffers:
+		port->io = data;
+		break;
+	default:
+		return -ENOENT;
+	}
+	return 0;
+}
+
+static int impl_node_port_reuse_buffer(void *object, uint32_t port_id, uint32_t buffer_id)
+{
+	return -ENOTSUP;
+}
+
+static int impl_node_process(void *object)
+{
+	struct impl *this = object;
+	struct port *port;
+	struct spa_io_buffers *io;
+	uint64_t now_time;
+
+	spa_return_val_if_fail(this != NULL, -EINVAL);
+
+	this = SPA_CONTAINER_OF(this, struct impl, node);
+	port = &this->port;
+	io = port->io;
+	spa_return_val_if_fail(io != NULL, -EIO);
+
+	spa_system_clock_gettime(this->data_system, CLOCK_MONOTONIC, &this->now);
+	now_time = SPA_TIMESPEC_TO_NSEC(&this->now);
+
+	if (!spa_list_is_empty(&port->ready))
+		render_buffers(this, now_time);
+
+	if (io->status == SPA_STATUS_HAVE_BUFFER && io->buffer_id < port->n_buffers) {
+		struct buffer *b = &port->buffers[io->buffer_id];
+
+		if (!b->outstanding) {
+			spa_log_warn(this->log, NAME " %p: buffer %u in use", this, io->buffer_id);
+			io->status = -EINVAL;
+			return -EINVAL;
+		}
+
+		spa_log_trace(this->log, NAME " %p: queue buffer %u", this, io->buffer_id);
+
+		spa_list_append(&port->ready, &b->link);
+		b->outstanding = false;
+		port->need_data = false;
+
+		this->threshold = SPA_MIN(b->buf->datas[0].chunk->size / port->frame_size,
+				this->props.max_latency);
+
+		render_buffers(this, now_time);
+
+		io->status = SPA_STATUS_OK;
+	}
+
+	return SPA_STATUS_HAVE_BUFFER;
+}
+
+static const struct spa_node_methods impl_node = {
+	SPA_VERSION_NODE_METHODS,
+	.add_listener = impl_node_add_listener,
+	.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,
+	.add_port = impl_node_add_port,
+	.remove_port = impl_node_remove_port,
+	.port_enum_params = impl_node_port_enum_params,
+	.port_set_param = impl_node_port_set_param,
+	.port_use_buffers = impl_node_port_use_buffers,
+	.port_set_io = impl_node_port_set_io,
+	.port_reuse_buffer = impl_node_port_reuse_buffer,
+	.process = impl_node_process,
+};
+
+static void transport_destroy(void *data)
+{
+	struct impl *this = data;
+	spa_log_debug(this->log, "transport %p destroy", this->transport);
+	this->transport = NULL;
+}
+
+static const struct spa_bt_transport_events transport_events = {
+	SPA_VERSION_BT_TRANSPORT_EVENTS,
+        .destroy = transport_destroy,
+};
+
+static int impl_get_interface(struct spa_handle *handle, uint32_t type, void **interface)
+{
+	struct impl *this;
+
+	spa_return_val_if_fail(handle != NULL, -EINVAL);
+	spa_return_val_if_fail(interface != NULL, -EINVAL);
+
+	this = (struct impl *) handle;
+
+	if (type == SPA_TYPE_INTERFACE_Node)
+		*interface = &this->node;
+	else
+		return -ENOENT;
+
+	return 0;
+}
+
+static int impl_clear(struct spa_handle *handle)
+{
+	struct impl *this = (struct impl *) handle;
+	spa_system_close(this->data_system, this->timerfd);
+	return 0;
+}
+
+static size_t
+impl_get_size(const struct spa_handle_factory *factory,
+	      const struct spa_dict *params)
+{
+	return sizeof(struct impl);
+}
+
+static int
+impl_init(const struct spa_handle_factory *factory,
+	  struct spa_handle *handle,
+	  const struct spa_dict *info,
+	  const struct spa_support *support,
+	  uint32_t n_support)
+{
+	struct impl *this;
+	struct port *port;
+	uint32_t i;
+
+	spa_return_val_if_fail(factory != NULL, -EINVAL);
+	spa_return_val_if_fail(handle != NULL, -EINVAL);
+
+	handle->get_interface = impl_get_interface;
+	handle->clear = impl_clear;
+
+	this = (struct impl *) handle;
+
+	for (i = 0; i < n_support; i++) {
+		switch (support[i].type) {
+		case SPA_TYPE_INTERFACE_Log:
+			this->log = support[i].data;
+			break;
+		case SPA_TYPE_INTERFACE_DataLoop:
+			this->data_loop = support[i].data;
+			break;
+		case SPA_TYPE_INTERFACE_DataSystem:
+			this->data_system = support[i].data;
+			break;
+		}
+	}
+	if (this->data_loop == NULL) {
+		spa_log_error(this->log, "a data loop is needed");
+		return -EINVAL;
+	}
+	if (this->data_system == NULL) {
+		spa_log_error(this->log, "a data system is needed");
+		return -EINVAL;
+	}
+
+	this->node.iface = SPA_INTERFACE_INIT(
+			SPA_TYPE_INTERFACE_Node,
+			SPA_VERSION_NODE,
+			&impl_node, this);
+	spa_hook_list_init(&this->hooks);
+
+	reset_props(&this->props);
+
+	this->info_all = SPA_NODE_CHANGE_MASK_FLAGS |
+			SPA_NODE_CHANGE_MASK_PARAMS |
+			SPA_NODE_CHANGE_MASK_PROPS;
+	this->info = SPA_NODE_INFO_INIT();
+	this->info.max_input_ports = 1;
+	this->info.max_output_ports = 0;
+	this->info.flags = SPA_NODE_FLAG_RT;
+	this->params[0] = SPA_PARAM_INFO(SPA_PARAM_PropInfo, SPA_PARAM_INFO_READ);
+	this->params[1] = SPA_PARAM_INFO(SPA_PARAM_Props, SPA_PARAM_INFO_READWRITE);
+	this->info.params = this->params;
+	this->info.n_params = 2;
+
+	port = &this->port;
+	port->info_all = SPA_PORT_CHANGE_MASK_FLAGS |
+			SPA_PORT_CHANGE_MASK_PARAMS;
+	port->info = SPA_PORT_INFO_INIT();
+	port->info.flags = 0;
+	port->params[0] = SPA_PARAM_INFO(SPA_PARAM_EnumFormat, SPA_PARAM_INFO_READ);
+	port->params[1] = SPA_PARAM_INFO(SPA_PARAM_Meta, SPA_PARAM_INFO_READ);
+	port->params[2] = SPA_PARAM_INFO(SPA_PARAM_IO, SPA_PARAM_INFO_READ);
+	port->params[3] = SPA_PARAM_INFO(SPA_PARAM_Format, SPA_PARAM_INFO_WRITE);
+	port->params[4] = SPA_PARAM_INFO(SPA_PARAM_Buffers, 0);
+	port->info.params = port->params;
+	port->info.n_params = 5;
+	spa_list_init(&port->ready);
+
+	for (i = 0; info && i < info->n_items; i++) {
+		if (strcmp(info->items[i].key, SPA_KEY_API_BLUEZ5_TRANSPORT) == 0)
+			sscanf(info->items[i].value, "pointer:%p", &this->transport);
+	}
+	if (this->transport == NULL) {
+		spa_log_error(this->log, "a transport is needed");
+		return -EINVAL;
+	}
+	spa_bt_transport_add_listener(this->transport,
+			&this->transport_listener, &transport_events, this);
+	this->sock_fd = -1;
+
+	this->timerfd = spa_system_timerfd_create(this->data_system,
+			CLOCK_MONOTONIC, SPA_FD_CLOEXEC | SPA_FD_NONBLOCK);
+
+	return 0;
+}
+
+static const struct spa_interface_info impl_interfaces[] = {
+	{SPA_TYPE_INTERFACE_Node,},
+};
+
+static int
+impl_enum_interface_info(const struct spa_handle_factory *factory,
+			 const struct spa_interface_info **info, uint32_t *index)
+{
+	spa_return_val_if_fail(factory != NULL, -EINVAL);
+	spa_return_val_if_fail(info != NULL, -EINVAL);
+	spa_return_val_if_fail(index != NULL, -EINVAL);
+
+	switch (*index) {
+	case 0:
+		*info = &impl_interfaces[*index];
+		break;
+	default:
+		return 0;
+	}
+	(*index)++;
+	return 1;
+}
+
+static const struct spa_dict_item info_items[] = {
+	{ SPA_KEY_FACTORY_AUTHOR, "Collabora Ltd. <contact@collabora.com>" },
+	{ SPA_KEY_FACTORY_DESCRIPTION, "Play bluetooth audio with hsp/hfp" },
+	{ SPA_KEY_FACTORY_USAGE, SPA_KEY_API_BLUEZ5_TRANSPORT"=<transport>" },
+};
+
+static const struct spa_dict info = SPA_DICT_INIT_ARRAY(info_items);
+
+struct spa_handle_factory spa_sco_sink_factory = {
+	SPA_VERSION_HANDLE_FACTORY,
+	SPA_NAME_API_BLUEZ5_SCO_SINK,
+	&info,
+	impl_get_size,
+	impl_init,
+	impl_enum_interface_info,
+};
diff --git a/spa/plugins/bluez5/sco-source.c b/spa/plugins/bluez5/sco-source.c
new file mode 100644
index 00000000..225415cb
--- /dev/null
+++ b/spa/plugins/bluez5/sco-source.c
@@ -0,0 +1,1116 @@
+/* Spa SCO Source
+ *
+ * Copyright © 2019 Collabora Ltd.
+ *
+ * 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 <unistd.h>
+#include <stddef.h>
+#include <stdio.h>
+#include <time.h>
+#include <fcntl.h>
+#include <sys/types.h>
+#include <sys/socket.h>
+
+#include <spa/support/loop.h>
+#include <spa/support/log.h>
+#include <spa/support/system.h>
+#include <spa/utils/list.h>
+#include <spa/utils/keys.h>
+#include <spa/utils/names.h>
+#include <spa/monitor/device.h>
+
+#include <spa/node/node.h>
+#include <spa/node/utils.h>
+#include <spa/node/io.h>
+#include <spa/param/param.h>
+#include <spa/param/audio/format.h>
+#include <spa/param/audio/format-utils.h>
+#include <spa/pod/filter.h>
+
+#include "defs.h"
+
+struct props {
+	uint32_t min_latency;
+	uint32_t max_latency;
+};
+
+#define MAX_BUFFERS 32
+
+struct buffer {
+	uint32_t id;
+	unsigned int outstanding:1;
+	struct spa_buffer *buf;
+	struct spa_meta_header *h;
+	struct spa_list link;
+};
+
+struct port {
+	struct spa_audio_info current_format;
+	int frame_size;
+	unsigned int have_format:1;
+
+	uint64_t info_all;
+	struct spa_port_info info;
+	struct spa_io_buffers *io;
+	struct spa_param_info params[8];
+
+	struct buffer buffers[MAX_BUFFERS];
+	uint32_t n_buffers;
+
+	struct spa_list free;
+	struct spa_list ready;
+
+	size_t ready_offset;
+};
+
+struct impl {
+	struct spa_handle handle;
+	struct spa_node node;
+
+	struct spa_log *log;
+	struct spa_loop *data_loop;
+	struct spa_system *data_system;
+
+	struct spa_hook_list hooks;
+	struct spa_callbacks callbacks;
+
+	uint64_t info_all;
+	struct spa_node_info info;
+	struct spa_param_info params[8];
+	struct props props;
+
+	struct spa_bt_transport *transport;
+	struct spa_hook transport_listener;
+	int sock_fd;
+
+	struct port port;
+
+	unsigned int started:1;
+	unsigned int slaved:1;
+
+	struct spa_source source;
+
+	struct spa_io_clock *clock;
+        struct spa_io_position *position;
+
+	struct timespec now;
+	uint32_t sample_count;
+};
+
+#define NAME "sco-source"
+
+#define CHECK_PORT(this,d,p)    ((d) == SPA_DIRECTION_OUTPUT && (p) == 0)
+
+static const uint32_t default_min_latency = 64;
+static const uint32_t default_max_latency = 256;
+
+static void reset_props(struct props *props)
+{
+	props->min_latency = default_min_latency;
+	props->max_latency = default_max_latency;
+}
+
+static int impl_node_enum_params(void *object, int seq,
+			uint32_t id, uint32_t start, uint32_t num,
+			const struct spa_pod *filter)
+{
+	struct impl *this = object;
+	struct spa_pod *param;
+	struct spa_pod_builder b = { 0 };
+	uint8_t buffer[1024];
+	struct spa_result_node_params result;
+	uint32_t count = 0;
+
+	spa_return_val_if_fail(this != NULL, -EINVAL);
+	spa_return_val_if_fail(num != 0, -EINVAL);
+
+        result.id = id;
+	result.next = start;
+      next:
+        result.index = result.next++;
+
+	spa_pod_builder_init(&b, buffer, sizeof(buffer));
+
+	switch (id) {
+	case SPA_PARAM_PropInfo:
+	{
+		struct props *p = &this->props;
+
+		switch (result.index) {
+		case 0:
+			param = spa_pod_builder_add_object(&b,
+				SPA_TYPE_OBJECT_PropInfo, id,
+				SPA_PROP_INFO_id,   SPA_POD_Id(SPA_PROP_minLatency),
+				SPA_PROP_INFO_name, SPA_POD_String("The minimum latency"),
+				SPA_PROP_INFO_type, SPA_POD_CHOICE_RANGE_Int(p->min_latency, 1, INT32_MAX));
+			break;
+		case 1:
+			param = spa_pod_builder_add_object(&b,
+				SPA_TYPE_OBJECT_PropInfo, id,
+				SPA_PROP_INFO_id,   SPA_POD_Id(SPA_PROP_maxLatency),
+				SPA_PROP_INFO_name, SPA_POD_String("The maximum latency"),
+				SPA_PROP_INFO_type, SPA_POD_CHOICE_RANGE_Int(p->max_latency, 1, INT32_MAX));
+			break;
+		default:
+			return 0;
+		}
+		break;
+	}
+	case SPA_PARAM_Props:
+	{
+		struct props *p = &this->props;
+
+		switch (result.index) {
+		case 0:
+			param = spa_pod_builder_add_object(&b,
+				SPA_TYPE_OBJECT_Props, id,
+				SPA_PROP_minLatency, SPA_POD_Int(p->min_latency),
+				SPA_PROP_maxLatency, SPA_POD_Int(p->max_latency));
+			break;
+		default:
+			return 0;
+		}
+		break;
+	}
+	default:
+		return -ENOENT;
+	}
+
+	if (spa_pod_filter(&b, &result.param, param, filter) < 0)
+		goto next;
+
+	spa_node_emit_result(&this->hooks, seq, 0, SPA_RESULT_TYPE_NODE_PARAMS, &result);
+
+	if (++count != num)
+		goto next;
+
+	return 0;
+}
+
+static int do_reslave(struct spa_loop *loop,
+			bool async,
+			uint32_t seq,
+			const void *data,
+			size_t size,
+			void *user_data)
+{
+	return 0;
+}
+
+static inline bool is_slaved(struct impl *this)
+{
+	return this->position && this->clock && this->position->clock.id != this->clock->id;
+}
+
+static int impl_node_set_io(void *object, uint32_t id, void *data, size_t size)
+{
+	struct impl *this = object;
+	bool slaved;
+
+	spa_return_val_if_fail(this != NULL, -EINVAL);
+
+	switch (id) {
+	case SPA_IO_Clock:
+		this->clock = data;
+		break;
+	case SPA_IO_Position:
+		this->position = data;
+		break;
+	default:
+		return -ENOENT;
+	}
+
+	slaved = is_slaved(this);
+	if (this->started && slaved != this->slaved) {
+		spa_log_debug(this->log, "sco-source %p: reslave %d->%d", this, this->slaved, slaved);
+		this->slaved = slaved;
+		spa_loop_invoke(this->data_loop, do_reslave, 0, NULL, 0, true, this);
+	}
+	return 0;
+}
+
+static int impl_node_set_param(void *object, uint32_t id, uint32_t flags,
+			       const struct spa_pod *param)
+{
+	struct impl *this = object;
+
+	spa_return_val_if_fail(this != NULL, -EINVAL);
+
+	switch (id) {
+	case SPA_PARAM_Props:
+	{
+		struct props *p = &this->props;
+
+		if (param == NULL) {
+			reset_props(p);
+			return 0;
+		}
+		spa_pod_parse_object(param,
+			SPA_TYPE_OBJECT_Props, NULL,
+			SPA_PROP_minLatency, SPA_POD_OPT_Int(&p->min_latency),
+			SPA_PROP_maxLatency, SPA_POD_OPT_Int(&p->max_latency));
+		break;
+	}
+	default:
+		return -ENOENT;
+	}
+
+	return 0;
+}
+
+static void reset_buffers(struct port *port)
+{
+	uint32_t i;
+
+	spa_list_init(&port->free);
+	spa_list_init(&port->ready);
+
+	for (i = 0; i < port->n_buffers; i++) {
+		struct buffer *b = &port->buffers[i];
+		spa_list_append(&port->free, &b->link);
+		b->outstanding = false;
+	}
+}
+
+static bool read_data(struct impl *this, uint8_t *data, uint32_t size, uint32_t *total_read) {
+	const uint32_t mtu_size = this->transport->read_mtu;
+	uint32_t local_total_read = 0;
+
+	/* TODO: For now we assume the size is always a mutliple of mtu_size */
+	while (local_total_read < (size - mtu_size)) {
+		const int bytes_read = read(this->sock_fd, data, mtu_size);
+		if (bytes_read == 0) {
+			/* Stop */
+			return false;
+		} else if (bytes_read < 0) {
+			/* Retry */
+			if (errno == EINTR)
+				continue;
+
+			/* Socked has no data so return total data read */
+			if (errno == EAGAIN || errno == EWOULDBLOCK)
+			    goto done;
+
+			/* Print error and stop */
+			spa_log_error(this->log, "read error: %s", strerror(errno));
+			return false;
+		}
+
+		data += bytes_read;
+		local_total_read += bytes_read;
+	}
+
+done:
+	if (total_read)
+	  *total_read = local_total_read;
+	return true;
+}
+
+static void sco_on_ready_read(struct spa_source *source)
+{
+	struct impl *this = source->data;
+	struct port *port = &this->port;
+	struct buffer *buffer;
+	struct spa_data *buffer_data;
+	uint32_t total_read;
+
+	/* update the current pts */
+	clock_gettime(CLOCK_MONOTONIC, &this->now);
+
+	/* check if we have a new buffer */
+	if (spa_list_is_empty(&port->free)) {
+		spa_log_warn(this->log, "waiting for buffer");
+		return;
+	}
+
+	/* get the buffer data */
+	buffer = spa_list_first(&port->free, struct buffer, link);
+	buffer_data = &buffer->buf->datas[0];
+	spa_assert(buffer_data->data);
+
+	/* read data */
+	if (!read_data(this, buffer_data->data, buffer_data->maxsize, &total_read))
+		goto stop;
+	if (total_read == 0)
+		return;
+
+	/* update the buffer offset, size and stride */
+	buffer_data->chunk->offset = 0;
+	buffer_data->chunk->size = total_read;
+	buffer_data->chunk->stride = port->frame_size;
+
+	/* update the sample count */
+	this->sample_count += buffer_data->chunk->size / port->frame_size;
+
+        /* remove the buffer from the free list and add it to the ready list */
+        spa_list_remove(&buffer->link);
+        buffer->outstanding = true;
+        spa_list_append(&port->ready, &buffer->link);
+
+        /* Notify we are ready for the next buffer */
+        spa_node_call_ready(&this->callbacks, SPA_STATUS_HAVE_BUFFER);
+
+	return;
+
+stop:
+	if (this->source.loop)
+		spa_loop_remove_source(this->data_loop, &this->source);
+}
+
+static int do_start(struct impl *this)
+{
+	int val;
+	bool do_accept;
+
+	/* Dont do anything if the node has already started */
+	if (this->started)
+		return 0;
+
+	/* Make sure the transport is valid */
+	spa_return_val_if_fail (this->transport != NULL, -EIO);
+
+	/* Do accept if Gateway; otherwise do connect for Head Unit */
+	do_accept = this->transport->profile & SPA_BT_PROFILE_HEADSET_AUDIO_GATEWAY;
+
+	/* acquire the socked fd (false -> connect | true -> accept) */
+	this->sock_fd = spa_bt_transport_acquire(this->transport, do_accept);
+	if (this->sock_fd < 0)
+		return -1;
+
+	/* Set the write MTU */
+	val = this->transport->write_mtu;
+	if (setsockopt(this->sock_fd, SOL_SOCKET, SO_SNDBUF, &val, sizeof(val)) < 0)
+		spa_log_warn(this->log, "sco-source %p: SO_SNDBUF %m", this);
+
+	/* Set the read MTU */
+	val = this->transport->read_mtu;
+	if (setsockopt(this->sock_fd, SOL_SOCKET, SO_RCVBUF, &val, sizeof(val)) < 0)
+		spa_log_warn(this->log, "sco-source %p: SO_RCVBUF %m", this);
+
+	/* Set the priority */
+	val = 6;
+	if (setsockopt(this->sock_fd, SOL_SOCKET, SO_PRIORITY, &val, sizeof(val)) < 0)
+		spa_log_warn(this->log, "SO_PRIORITY failed: %m");
+
+	/* Reset the buffers and sample count */
+	reset_buffers(&this->port);
+	this->sample_count = 0;
+
+	/* Add the ready read callback */
+	this->source.data = this;
+	this->source.fd = this->sock_fd;
+	this->source.func = sco_on_ready_read;
+	this->source.mask = SPA_IO_IN;
+	this->source.rmask = 0;
+	spa_loop_add_source(this->data_loop, &this->source);
+
+	/* Set the started flag */
+	this->started = true;
+
+	return 0;
+}
+
+static int do_remove_source(struct spa_loop *loop,
+			    bool async,
+			    uint32_t seq,
+			    const void *data,
+			    size_t size,
+			    void *user_data)
+{
+	struct impl *this = user_data;
+
+	if (this->source.loop)
+		spa_loop_remove_source(this->data_loop, &this->source);
+
+	return 0;
+}
+
+static int do_stop(struct impl *this)
+{
+	int res = 0;
+
+	if (!this->started)
+		return 0;
+
+	spa_log_debug(this->log, "sco-source %p: stop", this);
+
+	spa_loop_invoke(this->data_loop, do_remove_source, 0, NULL, 0, true, this);
+
+	this->started = false;
+
+	if (this->transport) {
+		/* Release the transport */
+		res = spa_bt_transport_release(this->transport);
+
+		/* Shutdown and close the socket */
+		shutdown(this->sock_fd, SHUT_RDWR);
+		close(this->sock_fd);
+		this->sock_fd = -1;
+	}
+
+	return res;
+}
+
+static int impl_node_send_command(void *object, const struct spa_command *command)
+{
+	struct impl *this = object;
+	struct port *port;
+	int res;
+
+	spa_return_val_if_fail(this != NULL, -EINVAL);
+	spa_return_val_if_fail(command != NULL, -EINVAL);
+
+	port = &this->port;
+
+	switch (SPA_NODE_COMMAND_ID(command)) {
+	case SPA_NODE_COMMAND_Start:
+		if (!port->have_format)
+			return -EIO;
+		if (port->n_buffers == 0)
+			return -EIO;
+
+		if ((res = do_start(this)) < 0)
+			return res;
+		break;
+	case SPA_NODE_COMMAND_Pause:
+		if ((res = do_stop(this)) < 0)
+			return res;
+		break;
+	default:
+		return -ENOTSUP;
+	}
+	return 0;
+}
+
+static const struct spa_dict_item node_info_items[] = {
+	{ SPA_KEY_DEVICE_API, "bluez5" },
+	{ SPA_KEY_MEDIA_CLASS, "Audio/Source" },
+	{ SPA_KEY_NODE_DRIVER, "true" },
+};
+
+static void emit_node_info(struct impl *this, bool full)
+{
+	if (full)
+		this->info.change_mask = this->info_all;
+	if (this->info.change_mask) {
+		this->info.props = &SPA_DICT_INIT_ARRAY(node_info_items);
+		spa_node_emit_info(&this->hooks, &this->info);
+		this->info.change_mask = 0;
+	}
+}
+
+static void emit_port_info(struct impl *this, struct port *port, bool full)
+{
+	if (full)
+		port->info.change_mask = port->info_all;
+	if (port->info.change_mask) {
+		spa_node_emit_port_info(&this->hooks,
+				SPA_DIRECTION_OUTPUT, 0, &port->info);
+		port->info.change_mask = 0;
+	}
+}
+
+static int
+impl_node_add_listener(void *object,
+		struct spa_hook *listener,
+		const struct spa_node_events *events,
+		void *data)
+{
+	struct impl *this = object;
+	struct spa_hook_list save;
+
+	spa_return_val_if_fail(this != NULL, -EINVAL);
+
+	spa_hook_list_isolate(&this->hooks, &save, listener, events, data);
+
+	emit_node_info(this, true);
+	emit_port_info(this, &this->port, true);
+
+	spa_hook_list_join(&this->hooks, &save);
+
+	return 0;
+}
+
+static int
+impl_node_set_callbacks(void *object,
+			const struct spa_node_callbacks *callbacks,
+			void *data)
+{
+	struct impl *this = object;
+
+	spa_return_val_if_fail(this != NULL, -EINVAL);
+
+	this->callbacks = SPA_CALLBACKS_INIT(callbacks, data);
+
+	return 0;
+}
+
+static int impl_node_sync(void *object, int seq)
+{
+	struct impl *this = object;
+
+	spa_return_val_if_fail(this != NULL, -EINVAL);
+
+	spa_node_emit_result(&this->hooks, seq, 0, 0, NULL);
+
+	return 0;
+}
+
+static int impl_node_add_port(void *object, enum spa_direction direction, uint32_t port_id,
+		const struct spa_dict *props)
+{
+	return -ENOTSUP;
+}
+
+static int impl_node_remove_port(void *object, enum spa_direction direction, uint32_t port_id)
+{
+	return -ENOTSUP;
+}
+
+static int
+impl_node_port_enum_params(void *object, int seq,
+			enum spa_direction direction, uint32_t port_id,
+			uint32_t id, uint32_t start, uint32_t num,
+			const struct spa_pod *filter)
+{
+
+	struct impl *this = object;
+	struct port *port;
+	struct spa_pod *param;
+	struct spa_pod_builder b = { 0 };
+	uint8_t buffer[1024];
+	struct spa_result_node_params result;
+	uint32_t count = 0;
+
+	spa_return_val_if_fail(this != NULL, -EINVAL);
+	spa_return_val_if_fail(num != 0, -EINVAL);
+
+	spa_return_val_if_fail(CHECK_PORT(this, direction, port_id), -EINVAL);
+	port = &this->port;
+
+	result.id = id;
+	result.next = start;
+      next:
+        result.index = result.next++;
+
+	spa_pod_builder_init(&b, buffer, sizeof(buffer));
+
+	switch (id) {
+	case SPA_PARAM_EnumFormat:
+		if (result.index > 0)
+			return 0;
+
+		if (this->transport == NULL)
+			return -EIO;
+
+		/* set the info structure */
+		struct spa_audio_info_raw info = { 0, };
+		info.format = SPA_AUDIO_FORMAT_S16;
+		info.channels = 1;
+		info.position[0] = SPA_AUDIO_CHANNEL_MONO;
+
+		 /* TODO: For now we only handle HSP profiles which has always CVSD format,
+		 * but we eventually need to support HFP that can have both CVSD and MSBC formats */
+
+		 /* CVSD format has a rate of 8kHz
+		  * MSBC format has a rate of 16kHz */
+		info.rate = 8000;
+
+		/* build the param */
+		param = spa_format_audio_raw_build(&b, id, &info);
+		break;
+
+	case SPA_PARAM_Format:
+		if (!port->have_format)
+			return -EIO;
+		if (result.index > 0)
+			return 0;
+
+		param = spa_format_audio_raw_build(&b, id, &port->current_format.info.raw);
+		break;
+
+	case SPA_PARAM_Buffers:
+		if (!port->have_format)
+			return -EIO;
+		if (result.index > 0)
+			return 0;
+
+		param = spa_pod_builder_add_object(&b,
+			SPA_TYPE_OBJECT_ParamBuffers, id,
+			/* 8 buffers are enough to make sure we always have one available when decoding */
+			SPA_PARAM_BUFFERS_buffers, SPA_POD_CHOICE_RANGE_Int(8, 8, MAX_BUFFERS),
+			SPA_PARAM_BUFFERS_blocks,  SPA_POD_Int(1),
+			SPA_PARAM_BUFFERS_size,    SPA_POD_Int(this->props.max_latency * port->frame_size),
+			SPA_PARAM_BUFFERS_stride,  SPA_POD_Int(port->frame_size),
+			SPA_PARAM_BUFFERS_align,   SPA_POD_Int(16));
+		break;
+
+	case SPA_PARAM_Meta:
+		switch (result.index) {
+		case 0:
+			param = spa_pod_builder_add_object(&b,
+				SPA_TYPE_OBJECT_ParamMeta, id,
+				SPA_PARAM_META_type, SPA_POD_Id(SPA_META_Header),
+				SPA_PARAM_META_size, SPA_POD_Int(sizeof(struct spa_meta_header)));
+			break;
+		default:
+			return 0;
+		}
+		break;
+
+	default:
+		return -ENOENT;
+	}
+
+	/* TODO: why filer is != NULL when linking it with sco-sink? */
+	/* if filter is null sco-source cannot be linked with sco-sink,
+	 * so for now we always pass NULL */
+	if (spa_pod_filter(&b, &result.param, param, NULL) < 0)
+		goto next;
+
+	spa_node_emit_result(&this->hooks, seq, 0, SPA_RESULT_TYPE_NODE_PARAMS, &result);
+
+	if (++count != num)
+		goto next;
+
+	return 0;
+}
+
+static int clear_buffers(struct impl *this, struct port *port)
+{
+	do_stop(this);
+	if (port->n_buffers > 0) {
+		spa_list_init(&port->free);
+		spa_list_init(&port->ready);
+		port->n_buffers = 0;
+	}
+	return 0;
+}
+
+static int port_set_format(struct impl *this, struct port *port,
+			   uint32_t flags,
+			   const struct spa_pod *format)
+{
+	int err;
+
+	if (format == NULL) {
+		spa_log_info(this->log, "clear format");
+		clear_buffers(this, port);
+		port->have_format = false;
+	} else {
+		struct spa_audio_info info = { 0 };
+
+		if ((err = spa_format_parse(format, &info.media_type, &info.media_subtype)) < 0)
+			return err;
+
+		if (info.media_type != SPA_MEDIA_TYPE_audio ||
+		    info.media_subtype != SPA_MEDIA_SUBTYPE_raw)
+			return -EINVAL;
+
+		if (spa_format_audio_raw_parse(format, &info.info.raw) < 0)
+			return -EINVAL;
+
+		port->frame_size = info.info.raw.channels * 2;
+		port->current_format = info;
+		port->have_format = true;
+	}
+
+	port->info.change_mask |= SPA_PORT_CHANGE_MASK_PARAMS;
+	if (port->have_format) {
+		port->info.change_mask |= SPA_PORT_CHANGE_MASK_FLAGS;
+		port->info.flags = SPA_PORT_FLAG_LIVE;
+		port->info.change_mask |= SPA_PORT_CHANGE_MASK_RATE;
+		port->info.rate = SPA_FRACTION(1, port->current_format.info.raw.rate);
+		port->params[3] = SPA_PARAM_INFO(SPA_PARAM_Format, SPA_PARAM_INFO_READWRITE);
+		port->params[4] = SPA_PARAM_INFO(SPA_PARAM_Buffers, SPA_PARAM_INFO_READ);
+	} else {
+		port->params[3] = SPA_PARAM_INFO(SPA_PARAM_Format, SPA_PARAM_INFO_WRITE);
+		port->params[4] = SPA_PARAM_INFO(SPA_PARAM_Buffers, 0);
+	}
+	emit_port_info(this, port, false);
+
+	return 0;
+}
+
+static int
+impl_node_port_set_param(void *object,
+			 enum spa_direction direction, uint32_t port_id,
+			 uint32_t id, uint32_t flags,
+			 const struct spa_pod *param)
+{
+	struct impl *this = object;
+	struct port *port;
+	int res;
+
+	spa_return_val_if_fail(this != NULL, -EINVAL);
+
+	spa_return_val_if_fail(CHECK_PORT(node, direction, port_id), -EINVAL);
+	port = &this->port;
+
+	switch (id) {
+	case SPA_PARAM_Format:
+		res = port_set_format(this, port, flags, param);
+		break;
+	default:
+		res = -ENOENT;
+		break;
+	}
+	return res;
+}
+
+static int
+impl_node_port_use_buffers(void *object,
+			   enum spa_direction direction, uint32_t port_id,
+			   uint32_t flags,
+			   struct spa_buffer **buffers, uint32_t n_buffers)
+{
+	struct impl *this = object;
+	struct port *port;
+	uint32_t i;
+
+	spa_return_val_if_fail(this != NULL, -EINVAL);
+
+	spa_return_val_if_fail(CHECK_PORT(this, direction, port_id), -EINVAL);
+	port = &this->port;
+
+	spa_log_info(this->log, "use buffers %d", n_buffers);
+
+	if (!port->have_format)
+		return -EIO;
+
+	clear_buffers(this, port);
+
+	for (i = 0; i < n_buffers; i++) {
+		struct buffer *b = &port->buffers[i];
+		struct spa_data *d = buffers[i]->datas;
+
+		b->buf = buffers[i];
+		b->id = i;
+
+		b->h = spa_buffer_find_meta_data(buffers[i], SPA_META_Header, sizeof(*b->h));
+
+		if (!((d[0].type == SPA_DATA_MemFd ||
+		       d[0].type == SPA_DATA_DmaBuf ||
+		       d[0].type == SPA_DATA_MemPtr) && d[0].data != NULL)) {
+			spa_log_error(this->log, NAME " %p: need mapped memory", this);
+			return -EINVAL;
+		}
+		spa_list_append(&port->free, &b->link);
+		b->outstanding = false;
+	}
+	port->n_buffers = n_buffers;
+
+	return 0;
+}
+
+static int
+impl_node_port_set_io(void *object,
+		      enum spa_direction direction,
+		      uint32_t port_id,
+		      uint32_t id,
+		      void *data, size_t size)
+{
+	struct impl *this = object;
+	struct port *port;
+
+	spa_return_val_if_fail(this != NULL, -EINVAL);
+
+	spa_return_val_if_fail(CHECK_PORT(this, direction, port_id), -EINVAL);
+	port = &this->port;
+
+	switch (id) {
+	case SPA_IO_Buffers:
+		port->io = data;
+		break;
+	default:
+		return -ENOENT;
+	}
+	return 0;
+}
+
+static void recycle_buffer(struct impl *this, struct port *port, uint32_t buffer_id)
+{
+	struct buffer *b = &port->buffers[buffer_id];
+
+	if (b->outstanding) {
+		spa_log_trace(this->log, NAME " %p: recycle buffer %u", this, buffer_id);
+		spa_list_append(&port->free, &b->link);
+		b->outstanding = false;
+	}
+}
+
+static int impl_node_port_reuse_buffer(void *object, uint32_t port_id, uint32_t buffer_id)
+{
+	struct impl *this = object;
+	struct port *port;
+
+	spa_return_val_if_fail(this != NULL, -EINVAL);
+
+	spa_return_val_if_fail(port_id == 0, -EINVAL);
+	port = &this->port;
+
+	if (port->n_buffers == 0)
+		return -EIO;
+
+	if (buffer_id >= port->n_buffers)
+		return -EINVAL;
+
+	recycle_buffer(this, port, buffer_id);
+
+	return 0;
+}
+
+static int impl_node_process(void *object)
+{
+	struct impl *this = object;
+	struct port *port;
+	struct spa_io_buffers *io;
+	struct buffer *b;
+
+	/* get IO */
+	spa_return_val_if_fail(this != NULL, -EINVAL);
+
+	port = &this->port;
+	io = port->io;
+	spa_return_val_if_fail(io != NULL, -EIO);
+
+	/* don't do anything if IO does not need a buffer */
+	if (io->status != SPA_STATUS_NEED_BUFFER)
+		return io->status;
+
+	/* Recycle previously played buffer */
+	if (io->buffer_id != SPA_ID_INVALID &&
+	    io->buffer_id < port->n_buffers) {
+		spa_log_debug(this->log, "recycling buffer_id=%d", io->buffer_id);
+		recycle_buffer(this, port, io->buffer_id);
+		io->buffer_id = SPA_ID_INVALID;
+	}
+
+	/* Check if we have new buffers in the queue */
+	if (spa_list_is_empty(&port->ready))
+		return SPA_STATUS_HAVE_BUFFER;
+
+	/* Pop the new buffer from the queue */
+	b = spa_list_first(&port->ready, struct buffer, link);
+	spa_list_remove(&b->link);
+
+	/* Set the new buffer in IO to be played */
+	io->buffer_id = b->id;
+	io->status = SPA_STATUS_HAVE_BUFFER;
+
+	return SPA_STATUS_HAVE_BUFFER;
+}
+
+static const struct spa_node_methods impl_node = {
+	SPA_VERSION_NODE_METHODS,
+	.add_listener = impl_node_add_listener,
+	.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,
+	.add_port = impl_node_add_port,
+	.remove_port = impl_node_remove_port,
+	.port_enum_params = impl_node_port_enum_params,
+	.port_set_param = impl_node_port_set_param,
+	.port_use_buffers = impl_node_port_use_buffers,
+	.port_set_io = impl_node_port_set_io,
+	.port_reuse_buffer = impl_node_port_reuse_buffer,
+	.process = impl_node_process,
+};
+
+static void transport_destroy(void *data)
+{
+	struct impl *this = data;
+	spa_log_debug(this->log, "transport %p destroy", this->transport);
+	this->transport = NULL;
+}
+
+static const struct spa_bt_transport_events transport_events = {
+	SPA_VERSION_BT_TRANSPORT_EVENTS,
+        .destroy = transport_destroy,
+};
+
+static int impl_get_interface(struct spa_handle *handle, uint32_t type, void **interface)
+{
+	struct impl *this;
+
+	spa_return_val_if_fail(handle != NULL, -EINVAL);
+	spa_return_val_if_fail(interface != NULL, -EINVAL);
+
+	this = (struct impl *) handle;
+
+	if (type == SPA_TYPE_INTERFACE_Node)
+		*interface = &this->node;
+	else
+		return -ENOENT;
+
+	return 0;
+}
+
+static int impl_clear(struct spa_handle *handle)
+{
+	return 0;
+}
+
+static size_t
+impl_get_size(const struct spa_handle_factory *factory,
+	      const struct spa_dict *params)
+{
+	return sizeof(struct impl);
+}
+
+static int
+impl_init(const struct spa_handle_factory *factory,
+	  struct spa_handle *handle,
+	  const struct spa_dict *info,
+	  const struct spa_support *support,
+	  uint32_t n_support)
+{
+	struct impl *this;
+	struct port *port;
+	uint32_t i;
+
+	spa_return_val_if_fail(factory != NULL, -EINVAL);
+	spa_return_val_if_fail(handle != NULL, -EINVAL);
+
+	handle->get_interface = impl_get_interface;
+	handle->clear = impl_clear;
+
+	this = (struct impl *) handle;
+
+	for (i = 0; i < n_support; i++) {
+		switch (support[i].type) {
+		case SPA_TYPE_INTERFACE_Log:
+			this->log = support[i].data;
+			break;
+		case SPA_TYPE_INTERFACE_DataLoop:
+			this->data_loop = support[i].data;
+			break;
+		case SPA_TYPE_INTERFACE_DataSystem:
+			this->data_system = support[i].data;
+			break;
+		}
+	}
+	if (this->data_loop == NULL) {
+		spa_log_error(this->log, "a data loop is needed");
+		return -EINVAL;
+	}
+	if (this->data_system == NULL) {
+		spa_log_error(this->log, "a data system is needed");
+		return -EINVAL;
+	}
+
+	this->node.iface = SPA_INTERFACE_INIT(
+			SPA_TYPE_INTERFACE_Node,
+			SPA_VERSION_NODE,
+			&impl_node, this);
+	spa_hook_list_init(&this->hooks);
+
+	reset_props(&this->props);
+
+	/* set the node info */
+	this->info_all = SPA_NODE_CHANGE_MASK_FLAGS |
+			SPA_NODE_CHANGE_MASK_PROPS |
+			SPA_NODE_CHANGE_MASK_PARAMS;
+	this->info = SPA_NODE_INFO_INIT();
+	this->info.flags = SPA_NODE_FLAG_RT;
+	this->params[0] = SPA_PARAM_INFO(SPA_PARAM_PropInfo, SPA_PARAM_INFO_READ);
+	this->params[1] = SPA_PARAM_INFO(SPA_PARAM_Props, SPA_PARAM_INFO_READWRITE);
+	this->info.params = this->params;
+	this->info.n_params = 2;
+
+	/* set the port info */
+	port = &this->port;
+	port->info_all = SPA_PORT_CHANGE_MASK_FLAGS |
+			SPA_PORT_CHANGE_MASK_PARAMS;
+	port->info = SPA_PORT_INFO_INIT();
+	port->info.change_mask = SPA_PORT_CHANGE_MASK_FLAGS;
+	port->info.flags = SPA_PORT_FLAG_LIVE |
+			   SPA_PORT_FLAG_TERMINAL;
+	port->params[0] = SPA_PARAM_INFO(SPA_PARAM_EnumFormat, SPA_PARAM_INFO_READ);
+	port->params[1] = SPA_PARAM_INFO(SPA_PARAM_Meta, SPA_PARAM_INFO_READ);
+	port->params[2] = SPA_PARAM_INFO(SPA_PARAM_IO, SPA_PARAM_INFO_READ);
+	port->params[3] = SPA_PARAM_INFO(SPA_PARAM_Format, SPA_PARAM_INFO_WRITE);
+	port->params[4] = SPA_PARAM_INFO(SPA_PARAM_Buffers, 0);
+	port->info.params = port->params;
+	port->info.n_params = 5;
+
+	/* Init the buffer lists */
+	spa_list_init(&port->ready);
+	spa_list_init(&port->free);
+
+	for (i = 0; info && i < info->n_items; i++) {
+		if (strcmp(info->items[i].key, SPA_KEY_API_BLUEZ5_TRANSPORT) == 0)
+			sscanf(info->items[i].value, "pointer:%p", &this->transport);
+	}
+	if (this->transport == NULL) {
+		spa_log_error(this->log, "a transport is needed");
+		return -EINVAL;
+	}
+	spa_bt_transport_add_listener(this->transport,
+			&this->transport_listener, &transport_events, this);
+	this->sock_fd = -1;
+
+	return 0;
+}
+
+static const struct spa_interface_info impl_interfaces[] = {
+	{SPA_TYPE_INTERFACE_Node,},
+};
+
+static int
+impl_enum_interface_info(const struct spa_handle_factory *factory,
+			 const struct spa_interface_info **info, uint32_t *index)
+{
+	spa_return_val_if_fail(factory != NULL, -EINVAL);
+	spa_return_val_if_fail(info != NULL, -EINVAL);
+	spa_return_val_if_fail(index != NULL, -EINVAL);
+
+	switch (*index) {
+	case 0:
+		*info = &impl_interfaces[*index];
+		break;
+	default:
+		return 0;
+	}
+	(*index)++;
+	return 1;
+}
+
+static const struct spa_dict_item info_items[] = {
+	{ SPA_KEY_FACTORY_AUTHOR, "Collabora Ltd. <contact@collabora.com>" },
+	{ SPA_KEY_FACTORY_DESCRIPTION, "Capture bluetooth audio with hsp/hfp" },
+	{ SPA_KEY_FACTORY_USAGE, SPA_KEY_API_BLUEZ5_TRANSPORT"=<transport>" },
+};
+
+static const struct spa_dict info = SPA_DICT_INIT_ARRAY(info_items);
+
+struct spa_handle_factory spa_sco_source_factory = {
+	SPA_VERSION_HANDLE_FACTORY,
+	SPA_NAME_API_BLUEZ5_SCO_SOURCE,
+	&info,
+	impl_get_size,
+	impl_init,
+	impl_enum_interface_info,
+};