From fe11bfdba0213fa84f7f4985bd3795631ebf11c5 Mon Sep 17 00:00:00 2001 From: Pauli Virtanen Date: Sun, 30 Oct 2022 16:54:57 +0200 Subject: [PATCH] bluez5: implement Bluetooth MIDI nodes Implement BLE MIDI node, providing an input and output port. MIDI data received is normalized and produced on output port. Data from input port is converted to BLE MIDI and sent to the device. Event timestamps come from device clock, whose offset to wall clock we determine by correlating packet reception times with their timestamps. The jitter in packet reception is likely much larger than device clock error, so we average over that. --- spa/plugins/bluez5/meson.build | 1 + spa/plugins/bluez5/midi-node.c | 1907 ++++++++++++++++++++++++++++++++ spa/plugins/bluez5/plugin.c | 4 + 3 files changed, 1912 insertions(+) create mode 100644 spa/plugins/bluez5/midi-node.c diff --git a/spa/plugins/bluez5/meson.build b/spa/plugins/bluez5/meson.build index b44a8c4e6..ef7f7b1d0 100644 --- a/spa/plugins/bluez5/meson.build +++ b/spa/plugins/bluez5/meson.build @@ -33,6 +33,7 @@ bluez5_sources = [ 'dbus-monitor.c', 'midi-enum.c', 'midi-parser.c', + 'midi-node.c', ] bluez5_data = ['bluez-hardware.conf'] diff --git a/spa/plugins/bluez5/midi-node.c b/spa/plugins/bluez5/midi-node.c new file mode 100644 index 000000000..43563f665 --- /dev/null +++ b/spa/plugins/bluez5/midi-node.c @@ -0,0 +1,1907 @@ +/* Spa MIDI node + * + * Copyright © 2022 Pauli Virtanen + * + * 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 +#include +#include +#include +#include +#include +#include + +#include + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +#include +#include +#include +#include +#include +#include +#include +#include +#include + +#include "dbus-monitor.h" +#include "midi.h" + +static struct spa_log_topic log_topic = SPA_LOG_TOPIC(0, "spa.bluez5.midi.node"); +#undef SPA_LOG_TOPIC_DEFAULT +#define SPA_LOG_TOPIC_DEFAULT &log_topic + +#define DEFAULT_CLOCK_NAME "clock.system.monotonic" + +#define DLL_BW 0.05 + +#define DEFAULT_LATENCY_OFFSET (0 * SPA_NSEC_PER_MSEC) + +#define MAX_BUFFERS 32 + +#define MIDI_RINGBUF_SIZE (8192*4) + +struct props { + char clock_name[64]; + int64_t latency_offset; +}; + +struct midi_event_ringbuffer_entry { + uint64_t time; + unsigned int size; +}; + +struct midi_event_ringbuffer { + struct spa_ringbuffer rbuf; + uint8_t buf[MIDI_RINGBUF_SIZE]; +}; + +struct buffer { + uint32_t id; + unsigned int outgoing:1; + struct spa_buffer *buf; + struct spa_meta_header *h; + struct spa_list link; +}; + +struct time_sync { + uint64_t prev_recv_time; + uint64_t recv_time; + + uint16_t prev_device_timestamp; + uint16_t device_timestamp; + + uint64_t device_time; + + struct spa_dll dll; +}; + +struct port { + uint32_t id; + enum spa_direction direction; + + struct spa_audio_info current_format; + unsigned int have_format:1; + + uint64_t info_all; + struct spa_port_info info; + struct spa_io_buffers *io; + struct spa_latency_info latency; +#define IDX_EnumFormat 0 +#define IDX_Meta 1 +#define IDX_IO 2 +#define IDX_Format 3 +#define IDX_Buffers 4 +#define IDX_Latency 5 +#define N_PORT_PARAMS 6 + struct spa_param_info params[N_PORT_PARAMS]; + + struct buffer buffers[MAX_BUFFERS]; + uint32_t n_buffers; + + struct spa_list free; + struct spa_list ready; + + int fd; + uint16_t mtu; + + struct buffer *buffer; + struct spa_pod_builder builder; + struct spa_pod_frame frame; + + struct time_sync sync; + + unsigned int acquired:1; + DBusPendingCall *acquire_call; + + struct impl *impl; +}; + +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_dbus *dbus; + struct spa_dbus_connection *dbus_connection; + + DBusConnection *conn; + + struct spa_hook_list hooks; + struct spa_callbacks callbacks; + + uint64_t info_all; + struct spa_node_info info; +#define IDX_PropInfo 0 +#define IDX_Props 1 +#define IDX_NODE_IO 2 +#define N_NODE_PARAMS 3 + struct spa_param_info params[N_NODE_PARAMS]; + struct props props; + +#define PORT_IN 0 +#define PORT_OUT 1 +#define N_PORTS 2 + struct port ports[N_PORTS]; + + char *chr_path; + + unsigned int started:1; + unsigned int following:1; + + struct spa_source source; + struct spa_source timer_source; + + int timerfd; + + struct spa_io_clock *clock; + struct spa_io_position *position; + + uint32_t duration; + uint32_t rate; + + uint64_t current_time; + uint64_t next_time; + + struct midi_event_ringbuffer event_rbuf; + + struct spa_bt_midi_parser parser; + struct spa_bt_midi_parser tmp_parser; + uint8_t read_buffer[MIDI_MAX_MTU]; + + struct spa_bt_midi_writer writer; +}; + +#define CHECK_PORT(this,d,p) ((p) == 0 && ((d) == SPA_DIRECTION_INPUT || (d) == SPA_DIRECTION_OUTPUT)) +#define GET_PORT(this,d,p) (&(this)->ports[(d) == SPA_DIRECTION_OUTPUT ? PORT_OUT : PORT_IN]) + +static void midi_event_ringbuffer_init(struct midi_event_ringbuffer *mbuf) +{ + spa_ringbuffer_init(&mbuf->rbuf); +} + +static int midi_event_ringbuffer_push(struct midi_event_ringbuffer *mbuf, + uint64_t time, uint8_t *event, unsigned int size) +{ + const unsigned int bufsize = sizeof(mbuf->buf); + int32_t avail; + uint32_t index; + struct midi_event_ringbuffer_entry evt = { + .time = time, + .size = size + }; + + avail = spa_ringbuffer_get_write_index(&mbuf->rbuf, &index); + if (avail < 0 || avail + sizeof(evt) + size > bufsize) + return -ENOSPC; + + spa_ringbuffer_write_data(&mbuf->rbuf, mbuf->buf, bufsize, index % bufsize, + &evt, sizeof(evt)); + index += sizeof(evt); + spa_ringbuffer_write_update(&mbuf->rbuf, index); + spa_ringbuffer_write_data(&mbuf->rbuf, mbuf->buf, bufsize, index % bufsize, + event, size); + index += size; + spa_ringbuffer_write_update(&mbuf->rbuf, index); + + return 0; +} + +static int midi_event_ringbuffer_peek(struct midi_event_ringbuffer *mbuf, uint64_t *time, unsigned int *size) +{ + const unsigned bufsize = sizeof(mbuf->buf); + int32_t avail; + uint32_t index; + struct midi_event_ringbuffer_entry evt; + + avail = spa_ringbuffer_get_read_index(&mbuf->rbuf, &index); + if (avail < (int)sizeof(evt)) + return -ENOENT; + + spa_ringbuffer_read_data(&mbuf->rbuf, mbuf->buf, bufsize, index % bufsize, + &evt, sizeof(evt)); + + *time = evt.time; + *size = evt.size; + return 0; +} + +static int midi_event_ringbuffer_pop(struct midi_event_ringbuffer *mbuf, uint8_t *data, size_t max_size) +{ + const unsigned bufsize = sizeof(mbuf->buf); + int32_t avail; + uint32_t index; + struct midi_event_ringbuffer_entry evt; + + avail = spa_ringbuffer_get_read_index(&mbuf->rbuf, &index); + if (avail < (int)sizeof(evt)) + return -ENOENT; + + spa_ringbuffer_read_data(&mbuf->rbuf, mbuf->buf, bufsize, index % bufsize, + &evt, sizeof(evt)); + index += sizeof(evt); + avail -= sizeof(evt); + spa_ringbuffer_read_update(&mbuf->rbuf, index); + + if ((uint32_t)avail < evt.size) { + /* corrupted ringbuffer: should never happen */ + spa_assert_not_reached(); + return -EINVAL; + } + + if (evt.size <= max_size) + spa_ringbuffer_read_data(&mbuf->rbuf, mbuf->buf, bufsize, index % bufsize, + data, SPA_MIN(max_size, evt.size)); + index += evt.size; + spa_ringbuffer_read_update(&mbuf->rbuf, index); + + if (evt.size > max_size) + return -ENOSPC; + + return 0; +} + +static void reset_props(struct props *props) +{ + props->latency_offset = DEFAULT_LATENCY_OFFSET; + strncpy(props->clock_name, DEFAULT_CLOCK_NAME, sizeof(props->clock_name)); +} + +static bool is_following(struct impl *this) +{ + return this->position && this->clock && this->position->clock.id != this->clock->id; +} + +static int set_timeout(struct impl *this, uint64_t time) +{ + struct itimerspec ts; + ts.it_value.tv_sec = time / SPA_NSEC_PER_SEC; + ts.it_value.tv_nsec = time % SPA_NSEC_PER_SEC; + ts.it_interval.tv_sec = 0; + ts.it_interval.tv_nsec = 0; + return spa_system_timerfd_settime(this->data_system, + this->timerfd, SPA_FD_TIMER_ABSTIME, &ts, NULL); +} + +static int set_timers(struct impl *this) +{ + struct timespec now; + + spa_system_clock_gettime(this->data_system, CLOCK_MONOTONIC, &now); + this->next_time = SPA_TIMESPEC_TO_NSEC(&now); + + return set_timeout(this, this->following ? 0 : this->next_time); +} + +static void recycle_buffer(struct impl *this, struct port *port, uint32_t buffer_id) +{ + struct buffer *b = &port->buffers[buffer_id]; + + if (b->outgoing) { + spa_log_trace(this->log, "%p: recycle buffer %u", this, buffer_id); + spa_list_append(&port->free, &b->link); + b->outgoing = false; + } +} + +static int clear_buffers(struct impl *this, struct port *port) +{ + if (port->n_buffers > 0) { + spa_list_init(&port->free); + spa_list_init(&port->ready); + port->n_buffers = 0; + } + 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]; + + if (port->direction == SPA_DIRECTION_OUTPUT) { + spa_list_append(&port->free, &b->link); + b->outgoing = false; + } else { + b->outgoing = true; + } + } +} + +static struct buffer *peek_buffer(struct impl *this, struct port *port) +{ + if (spa_list_is_empty(&port->free)) + return NULL; + return spa_list_first(&port->free, struct buffer, link); +} + +static int prepare_buffer(struct impl *this, struct port *port) +{ + if (port->buffer != NULL) + return 0; + if ((port->buffer = peek_buffer(this, port)) == NULL) + return -EPIPE; + + spa_pod_builder_init(&port->builder, + port->buffer->buf->datas[0].data, + port->buffer->buf->datas[0].maxsize); + spa_pod_builder_push_sequence(&port->builder, &port->frame, 0); + + return 0; +} + +static int finish_buffer(struct impl *this, struct port *port) +{ + if (port->buffer == NULL) + return 0; + + spa_pod_builder_pop(&port->builder, &port->frame); + + port->buffer->buf->datas[0].chunk->offset = 0; + port->buffer->buf->datas[0].chunk->size = port->builder.state.offset; + + /* move buffer to ready queue */ + spa_list_remove(&port->buffer->link); + spa_list_append(&port->ready, &port->buffer->link); + port->buffer = NULL; + + return 0; +} + +/* Replace value -> value + n*period, to minimize |value - target| */ +static int64_t unwrap_to_closest(int64_t value, int64_t target, int64_t period) +{ + if (value > target) + value -= SPA_ROUND_DOWN(value - target + period/2, period); + if (value < target) + value += SPA_ROUND_DOWN(target - value + period/2, period); + return value; +} + +static int64_t time_diff(uint64_t a, uint64_t b) +{ + if (a >= b) + return a - b; + else + return -(int64_t)(b - a); +} + +static void midi_event_get_last_timestamp(void *user_data, uint16_t timestamp, uint8_t *data, size_t size) +{ + int *last_timestamp = user_data; + *last_timestamp = timestamp; +} + +static uint64_t midi_convert_time(struct time_sync *sync, uint16_t timestamp) +{ + int offset; + + /* + * sync->device_timestamp is a device timestamp that corresponds to system + * clock time sync->device_time. + * + * It is the timestamp of the last MIDI event in the current packet, so we can + * assume here no event here has timestamp after it. + */ + if (timestamp > sync->device_timestamp) + offset = sync->device_timestamp + MIDI_CLOCK_PERIOD_MSEC - timestamp; + else + offset = sync->device_timestamp - timestamp; + + return sync->device_time - offset * SPA_NSEC_PER_MSEC; +} + +static void midi_event_recv(void *user_data, uint16_t timestamp, uint8_t *data, size_t size) +{ + struct impl *this = user_data; + struct port *port = &this->ports[PORT_OUT]; + struct time_sync *sync = &port->sync; + uint64_t time; + int res; + + spa_assert(size > 0); + + time = midi_convert_time(sync, timestamp); + + spa_log_trace(this->log, "%p: event:0x%x size:%d timestamp:%d time:%"PRIu64"", + this, (int)data[0], (int)size, (int)timestamp, (uint64_t)time); + + res = midi_event_ringbuffer_push(&this->event_rbuf, time, data, size); + if (res < 0) { + midi_event_ringbuffer_init(&this->event_rbuf); + spa_log_warn(this->log, "%p: MIDI receive buffer overflow: %s", + this, spa_strerror(res)); + } +} + +static void on_ready_read(struct spa_source *source) +{ + struct impl *this = source->data; + struct port *port = &this->ports[PORT_OUT]; + struct timespec now; + int res, size, last_timestamp; + + spa_system_clock_gettime(this->data_system, CLOCK_MONOTONIC, &now); + + /* read data from socket */ +again: + size = recv(port->fd, this->read_buffer, sizeof(this->read_buffer), MSG_DONTWAIT | MSG_NOSIGNAL); + if (size == 0) { + return; + } else if (size < 0) { + if (errno == EINTR) + goto again; + if (errno == EAGAIN || errno == EWOULDBLOCK) + return; + goto stop; + } + + spa_log_trace(this->log, "%p: port:%d recv data size:%d", this, port->direction, size); + if (SPA_UNLIKELY(spa_log_level_topic_enabled(this->log, SPA_LOG_TOPIC_DEFAULT, SPA_LOG_LEVEL_TRACE))) + spa_log_hexdump(this->log, SPA_LOG_LEVEL_DEBUG, 4, this->read_buffer, size); + + /* prepare for producing events */ + if (port->io == NULL || port->n_buffers == 0 || !this->started) + return; + + /* + * Remote clock synchronization: + * + * Assume: Last timestamp in packet on average corresponds to packet send time. + * There is some unknown latency in between, but on average it is constant. + * + * The `device_time` computed below is the estimated wall-clock time + * corresponding to the timestamp `device_timestamp` of the last event + * in the packet. This timestamp is late by the average transmission latency, + * which is unknown. + * + * Packet reception jitter and any clock drift is smoothed over with DLL. + * The estimated timestamps are stable and preserve event intervals. + * + * To allow latency_offset to work better, we don't write the events + * to the output buffer here, but instead put them to a ringbuffer. + * This is because if the offset shifts events to later buffers, + * this is simpler to handle with the rbuf. + */ + last_timestamp = -1; + spa_bt_midi_parser_dup(&this->parser, &this->tmp_parser, true); + res = spa_bt_midi_parser_parse(&this->tmp_parser, this->read_buffer, size, true, + midi_event_get_last_timestamp, &last_timestamp); + if (res >= 0 && last_timestamp >= 0) { + struct time_sync *sync = &port->sync; + int64_t clock_elapsed; + int64_t device_elapsed; + int64_t err_nsec; + double corr, tcorr; + + sync->prev_recv_time = sync->recv_time; + sync->recv_time = SPA_TIMESPEC_TO_NSEC(&now); + + sync->prev_device_timestamp = sync->device_timestamp; + sync->device_timestamp = last_timestamp; + + if (port->sync.prev_recv_time == 0) { + sync->prev_recv_time = sync->recv_time; + sync->prev_device_timestamp = sync->device_timestamp; + spa_dll_init(&sync->dll); + } + if (SPA_UNLIKELY(sync->dll.bw == 0)) + spa_dll_set_bw(&sync->dll, DLL_BW, 1024, 48000); + + /* move device clock forward */ + clock_elapsed = sync->recv_time - sync->prev_recv_time; + + device_elapsed = (int)sync->device_timestamp - (int)sync->prev_device_timestamp; + device_elapsed *= SPA_NSEC_PER_MSEC; + device_elapsed = unwrap_to_closest(device_elapsed, clock_elapsed, MIDI_CLOCK_PERIOD_NSEC); + sync->device_time += device_elapsed; + + /* smooth clock sync */ + err_nsec = time_diff(sync->recv_time, sync->device_time); + corr = spa_dll_update(&sync->dll, + -SPA_CLAMP(err_nsec, -20*SPA_NSEC_PER_MSEC, 20*SPA_NSEC_PER_MSEC) + * this->rate / SPA_NSEC_PER_SEC); + tcorr = SPA_MIN(device_elapsed, SPA_NSEC_PER_SEC) * (corr - 1); + sync->device_time += tcorr; + + /* reset if too much off */ + if (err_nsec < -50 * SPA_NSEC_PER_MSEC || + err_nsec > 200 * SPA_NSEC_PER_MSEC || + SPA_ABS(tcorr) > 20*SPA_NSEC_PER_MSEC || + device_elapsed < 0) { + spa_log_debug(this->log, "%p: device clock sync off too much: resync", this); + spa_dll_init(&sync->dll); + sync->device_time = sync->recv_time; + } + + spa_log_debug(this->log, + "timestamp:%d dt:%d dt2:%d err:%.1f tcorr:%.2f (ms) corr:%f", + (int)sync->device_timestamp, + (int)(clock_elapsed/SPA_NSEC_PER_MSEC), + (int)(device_elapsed/SPA_NSEC_PER_MSEC), + (double)err_nsec / SPA_NSEC_PER_MSEC, + tcorr/SPA_NSEC_PER_MSEC, + corr); + } + + /* put midi event data to the buffer */ + res = spa_bt_midi_parser_parse(&this->parser, this->read_buffer, size, false, + midi_event_recv, this); + if (res < 0) { + /* bad data */ + spa_bt_midi_parser_init(&this->parser); + + spa_log_info(this->log, "BLE MIDI data packet parsing failed: %d", res); + if (spa_log_level_topic_enabled(this->log, SPA_LOG_TOPIC_DEFAULT, SPA_LOG_LEVEL_DEBUG)) + spa_log_hexdump(this->log, SPA_LOG_LEVEL_DEBUG, 4, this->read_buffer, size); + } + + return; + +stop: + if (this->source.loop) + spa_loop_remove_source(this->data_loop, &this->source); +} + +static int process_output(struct impl *this) +{ + struct port *port = &this->ports[PORT_OUT]; + struct buffer *buffer; + struct spa_io_buffers *io = port->io; + + /* Check if we are able to process */ + if (io == NULL || !port->acquired) + return SPA_STATUS_OK; + + /* Return if we already have a buffer */ + if (io->status == SPA_STATUS_HAVE_DATA) + return SPA_STATUS_HAVE_DATA; + + /* Recycle */ + if (io->buffer_id < port->n_buffers) { + recycle_buffer(this, port, io->buffer_id); + io->buffer_id = SPA_ID_INVALID; + } + + /* Produce buffer */ + if (prepare_buffer(this, port) >= 0) { + /* + * this->current_time is at the end time of the buffer, and offsets + * are recorded vs. the start of the buffer. + */ + const uint64_t start_time = this->current_time + - this->duration * SPA_NSEC_PER_SEC / this->rate; + const uint64_t end_time = this->current_time; + uint64_t time; + uint32_t offset; + void *buf; + unsigned int size; + int res; + + while (true) { + res = midi_event_ringbuffer_peek(&this->event_rbuf, &time, &size); + if (res < 0) + break; + + time -= this->props.latency_offset; + + if (time > end_time) { + break; + } else if (time + SPA_NSEC_PER_MSEC < start_time) { + /* Log events in the past by more than 1 ms, but don't + * do anything about them. The user can change the latency + * offset to choose whether to tradeoff latency for more + * accurate timestamps. + * + * TODO: maybe this information should be available in + * a more visible place, some latency property? + */ + spa_log_debug(this->log, "%p: event in the past by %d ms", + this, (int)((start_time - time) / SPA_NSEC_PER_MSEC)); + } + + time = SPA_MAX(time, start_time) - start_time; + offset = time * this->rate / SPA_NSEC_PER_SEC; + offset = SPA_CLAMP(offset, 0u, this->duration - 1); + + spa_pod_builder_control(&port->builder, offset, SPA_CONTROL_Midi); + buf = spa_pod_builder_reserve_bytes(&port->builder, size); + if (buf) { + midi_event_ringbuffer_pop(&this->event_rbuf, buf, size); + + spa_log_trace(this->log, "%p: produce event:0x%x offset:%d time:%"PRIu64"", + this, (int)*(uint8_t*)buf, (int)offset, + (uint64_t)(start_time + offset * SPA_NSEC_PER_SEC / this->rate)); + } + } + + finish_buffer(this, port); + } + + /* Return if there are no buffers ready to be processed */ + if (spa_list_is_empty(&port->ready)) + return SPA_STATUS_OK; + + /* Get the new buffer from the ready list */ + buffer = spa_list_first(&port->ready, struct buffer, link); + spa_list_remove(&buffer->link); + buffer->outgoing = true; + + /* Set the new buffer in IO */ + io->buffer_id = buffer->id; + io->status = SPA_STATUS_HAVE_DATA; + + /* Notify we have a buffer ready to be processed */ + return SPA_STATUS_HAVE_DATA; +} + +static int flush_packet(struct impl *this) +{ + struct port *port = &this->ports[PORT_IN]; + bool log_enabled = spa_log_level_topic_enabled(this->log, SPA_LOG_TOPIC_DEFAULT, SPA_LOG_LEVEL_TRACE); + int res; + + if (this->writer.size == 0) + return 0; + + res = send(port->fd, this->writer.buf, this->writer.size, + MSG_DONTWAIT | MSG_NOSIGNAL); + if (res < 0) + return -errno; + + spa_log_trace(this->log, "%p: send packet size:%d", this, this->writer.size); + if (log_enabled) + spa_log_hexdump(this->log, SPA_LOG_LEVEL_TRACE, 4, + this->writer.buf, this->writer.size); + + return 0; +} + +static int write_data(struct impl *this, struct spa_data *d) +{ + struct port *port = &this->ports[PORT_IN]; + struct spa_pod_sequence *pod; + struct spa_pod_control *c; + uint64_t time; + int res; + + pod = spa_pod_from_data(d->data, d->maxsize, d->chunk->offset, d->chunk->size); + if (pod == NULL) { + spa_log_warn(this->log, "%p: invalid sequence in buffer max:%u offset:%u size:%u", + this, d->maxsize, d->chunk->offset, d->chunk->size); + return -EINVAL; + } + + spa_bt_midi_writer_init(&this->writer, port->mtu); + time = 0; + + SPA_POD_SEQUENCE_FOREACH(pod, c) { + uint8_t *event; + size_t size; + + if (c->type != SPA_CONTROL_Midi) + continue; + + time = SPA_MAX(time, this->current_time + c->offset * SPA_NSEC_PER_SEC / this->rate); + event = SPA_POD_BODY(&c->value); + size = SPA_POD_BODY_SIZE(&c->value); + + spa_log_trace(this->log, "%p: output event:0x%x time:%"PRIu64, this, + (size > 0) ? event[0] : 0, time); + + do { + res = spa_bt_midi_writer_write(&this->writer, + time, event, size); + if (res < 0) { + return res; + } else if (res) { + int res2; + if ((res2 = flush_packet(this)) < 0) + return res2; + } + } while (res); + } + + if ((res = flush_packet(this)) < 0) + return res; + + return 0; +} + +static int process_input(struct impl *this) +{ + struct port *port = &this->ports[PORT_IN]; + struct buffer *b; + struct spa_io_buffers *io = port->io; + int res; + + /* Check if we are able to process */ + if (io == NULL || !port->acquired) + return SPA_STATUS_OK; + + if (io->status != SPA_STATUS_HAVE_DATA || io->buffer_id >= port->n_buffers) + return SPA_STATUS_OK; + + b = &port->buffers[io->buffer_id]; + if (!b->outgoing) { + spa_log_warn(this->log, "%p: buffer %u not outgoing", this, io->buffer_id); + io->status = -EINVAL; + return -EINVAL; + } + + if ((res = write_data(this, &b->buf->datas[0])) < 0) { + spa_log_info(this->log, "%p: writing data failed: %s", + this, spa_strerror(res)); + } + + port->io->buffer_id = b->id; + io->status = SPA_STATUS_NEED_DATA; + spa_node_call_reuse_buffer(&this->callbacks, 0, io->buffer_id); + + return SPA_STATUS_HAVE_DATA; +} + +static void update_position(struct impl *this) +{ + if (SPA_LIKELY(this->position)) { + this->duration = this->position->clock.duration; + this->rate = this->position->clock.rate.denom; + } else { + this->duration = 1024; + this->rate = 48000; + } +} + +static void on_timeout(struct spa_source *source) +{ + struct impl *this = source->data; + uint64_t exp; + uint64_t prev_time, now_time; + int status; + + if (!this->started) + return; + + if (spa_system_timerfd_read(this->data_system, this->timerfd, &exp) < 0) + spa_log_warn(this->log, "%p: error reading timerfd: %s", this, strerror(errno)); + + prev_time = this->current_time; + now_time = this->current_time = this->next_time; + + spa_log_trace(this->log, "%p: timer %"PRIu64" %"PRIu64"", this, + now_time, now_time - prev_time); + + update_position(this); + + this->next_time = now_time + this->duration * SPA_NSEC_PER_SEC / this->rate; + + if (SPA_LIKELY(this->clock)) { + this->clock->nsec = now_time; + this->clock->position += this->duration; + this->clock->duration = this->duration; + this->clock->rate_diff = 1.0f; + this->clock->next_nsec = this->next_time; + } + + status = process_output(this); + spa_log_trace(this->log, "%p: status:%d", this, status); + + spa_node_call_ready(&this->callbacks, status | SPA_STATUS_NEED_DATA); + + set_timeout(this, this->next_time); +} + +static int do_start(struct impl *this); + +static int do_release(struct impl *this); + +static int do_stop(struct impl *this); + +static void acquire_reply(DBusPendingCall **call_ptr, DBusMessage *r) +{ + struct port *port = SPA_CONTAINER_OF(call_ptr, struct port, acquire_call); + struct impl *this = port->impl; + const char *method = (port->direction == SPA_DIRECTION_OUTPUT) ? + "AcquireNotify" : "AcquireWrite"; + int fd; + uint16_t mtu; + + if (dbus_message_get_type(r) == DBUS_MESSAGE_TYPE_ERROR) { + spa_log_error(this->log, "%s.%s() failed for %s: %s", + BLUEZ_GATT_CHR_INTERFACE, method, + this->chr_path, dbus_message_get_error_name(r)); + do_release(this); + return; + } + + if (!dbus_message_get_args(r, NULL, + DBUS_TYPE_UNIX_FD, &fd, + DBUS_TYPE_UINT16, &mtu, + DBUS_TYPE_INVALID)) { + spa_log_error(this->log, "%s.%s for %s: invalid return value", + BLUEZ_GATT_CHR_INTERFACE, method, this->chr_path); + do_release(this); + return; + } + + spa_log_info(this->log, "%p: BLE MIDI %s %s success mtu:%d", + this, this->chr_path, method, mtu); + port->fd = fd; + port->mtu = mtu; + port->acquired = true; + + if (port->direction == SPA_DIRECTION_OUTPUT) { + spa_bt_midi_parser_init(&this->parser); + + update_position(this); + + /* Start source */ + this->source.data = this; + this->source.fd = port->fd; + this->source.func = on_ready_read; + this->source.mask = SPA_IO_IN; + this->source.rmask = 0; + spa_loop_add_source(this->data_loop, &this->source); + } +} + +static int do_acquire(struct port *port) +{ + struct impl *this = port->impl; + DBusMessageIter i, d; + DBusMessage *m; + const char *method = (port->direction == SPA_DIRECTION_OUTPUT) ? + "AcquireNotify" : "AcquireWrite"; + + if (port->acquired) + return 0; + if (port->acquire_call) + return 0; + + spa_log_debug(this->log, + "%p: acquiring BLE MIDI device characteristic %s", + this, this->chr_path); + + m = dbus_message_new_method_call(BLUEZ_SERVICE, + this->chr_path, + BLUEZ_GATT_CHR_INTERFACE, + method); + if (m == NULL) + return -EINVAL; + + dbus_message_iter_init_append(m, &i); + dbus_message_iter_open_container(&i, DBUS_TYPE_ARRAY, "{sv}", &d); + dbus_message_iter_close_container(&i, &d); + + return spa_dbus_async_call(this->conn, m, &port->acquire_call, + acquire_reply); +} + +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); + + if (this->timer_source.loop) + spa_loop_remove_source(this->data_loop, &this->timer_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; + spa_system_timerfd_settime(this->data_system, this->timerfd, 0, &ts, NULL); + + return 0; +} + +static int do_stop(struct impl *this) +{ + int res = 0; + + spa_log_debug(this->log, "%p: stop", this); + + spa_loop_invoke(this->data_loop, do_remove_source, 0, NULL, 0, true, this); + + this->started = false; + return res; +} + +static int do_release(struct impl *this) +{ + int res = 0; + size_t i; + + spa_log_debug(this->log, "%p: release", this); + + do_stop(this); + + for (i = 0; i < N_PORTS; ++i) { + struct port *port = &this->ports[i]; + + spa_dbus_async_call_cancel(&port->acquire_call); + + if (port->fd > 0) { + shutdown(port->fd, SHUT_RDWR); + close(port->fd); + port->fd = -1; + port->mtu = 0; + } + } + + return res; +} + +static int do_start(struct impl *this) +{ + int res; + size_t i; + + if (this->started) + return 0; + + this->following = is_following(this); + + update_position(this); + + spa_log_debug(this->log, "%p: start following:%d", + this, this->following); + + for (i = 0; i < N_PORTS; ++i) { + struct port *port = &this->ports[i]; + + /* Acquire Bluetooth I/O */ + if ((res = do_acquire(port)) < 0) { + do_release(this); + return res; + } + + reset_buffers(port); + } + + midi_event_ringbuffer_init(&this->event_rbuf); + + this->started = true; + + /* Start timer */ + this->timer_source.data = this; + this->timer_source.fd = this->timerfd; + this->timer_source.func = on_timeout; + this->timer_source.mask = SPA_IO_IN; + this->timer_source.rmask = 0; + spa_loop_add_source(this->data_loop, &this->timer_source); + + set_timers(this); + + return 0; +} + + +static int do_reassign_follower(struct spa_loop *loop, + bool async, + uint32_t seq, + const void *data, + size_t size, + void *user_data) +{ + struct impl *this = user_data; + + set_timers(this); + return 0; +} + +static int impl_node_set_io(void *object, uint32_t id, void *data, size_t size) +{ + struct impl *this = object; + bool following; + + spa_return_val_if_fail(this != NULL, -EINVAL); + + switch (id) { + case SPA_IO_Clock: + this->clock = data; + if (this->clock != NULL) { + spa_scnprintf(this->clock->name, + sizeof(this->clock->name), + "%s", this->props.clock_name); + } + break; + case SPA_IO_Position: + this->position = data; + break; + default: + return -ENOENT; + } + + following = is_following(this); + if (this->started && following != this->following) { + spa_log_debug(this->log, "%p: reassign follower %d->%d", this, this->following, following); + this->following = following; + spa_loop_invoke(this->data_loop, do_reassign_follower, 0, NULL, 0, true, this); + } + + return 0; +} + +static void emit_node_info(struct impl *this, bool full); + +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: + { + 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_latencyOffsetNsec), + SPA_PROP_INFO_description, SPA_POD_String("Latency offset (ns)"), + SPA_PROP_INFO_type, SPA_POD_CHOICE_RANGE_Long(0LL, INT64_MIN, INT64_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_latencyOffsetNsec, SPA_POD_Long(p->latency_offset)); + 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 emit_port_info(struct impl *this, struct port *port, bool full); + +static void set_latency(struct impl *this, bool emit_latency) +{ + struct port *port = &this->ports[PORT_OUT]; + + port->latency.min_ns = port->latency.max_ns = this->props.latency_offset; + + if (emit_latency) { + port->info.change_mask |= SPA_PORT_CHANGE_MASK_PARAMS; + port->params[IDX_Latency].flags ^= SPA_PARAM_INFO_SERIAL; + emit_port_info(this, port, false); + } +} + +static int apply_props(struct impl *this, const struct spa_pod *param) +{ + struct props new_props = this->props; + int changed = 0; + + if (param == NULL) { + reset_props(&new_props); + } else { + spa_pod_parse_object(param, + SPA_TYPE_OBJECT_Props, NULL, + SPA_PROP_latencyOffsetNsec, SPA_POD_OPT_Long(&new_props.latency_offset)); + } + + changed = (memcmp(&new_props, &this->props, sizeof(struct props)) != 0); + this->props = new_props; + + if (changed) + set_latency(this, true); + + return changed; +} + +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: + { + if (apply_props(this, param) > 0) { + this->info.change_mask |= SPA_NODE_CHANGE_MASK_PARAMS; + this->params[IDX_Props].flags ^= SPA_PARAM_INFO_SERIAL; + emit_node_info(this, false); + } + break; + } + default: + return -ENOENT; + } + + return 0; +} + +static int impl_node_send_command(void *object, const struct spa_command *command) +{ + struct impl *this = object; + int res; + + spa_return_val_if_fail(this != NULL, -EINVAL); + spa_return_val_if_fail(command != NULL, -EINVAL); + + switch (SPA_NODE_COMMAND_ID(command)) { + case SPA_NODE_COMMAND_Start: + if ((res = do_start(this)) < 0) + return res; + break; + case SPA_NODE_COMMAND_Pause: + case SPA_NODE_COMMAND_Suspend: + if ((res = do_stop(this)) < 0) + return res; + break; + default: + return -ENOTSUP; + } + return 0; +} + +static void emit_node_info(struct impl *this, bool full) +{ + const struct spa_dict_item node_info_items[] = { + { SPA_KEY_DEVICE_API, "bluez5" }, + { SPA_KEY_MEDIA_CLASS, "Midi/Bridge" }, + }; + uint64_t old = full ? this->info.change_mask : 0; + + 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 = old; + } +} + +static void emit_port_info(struct impl *this, struct port *port, bool full) +{ + uint64_t old = full ? port->info.change_mask : 0; + if (full) + port->info.change_mask = port->info_all; + if (port->info.change_mask) { + spa_node_emit_port_info(&this->hooks, port->direction, port->id, &port->info); + port->info.change_mask = old; + } +} + +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; + size_t i; + + spa_return_val_if_fail(this != NULL, -EINVAL); + + spa_hook_list_isolate(&this->hooks, &save, listener, events, data); + + emit_node_info(this, true); + + for (i = 0; i < N_PORTS; ++i) + emit_port_info(this, &this->ports[i], 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 = GET_PORT(this, direction, port_id); + + 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; + + param = spa_pod_builder_add_object(&b, + SPA_TYPE_OBJECT_Format, SPA_PARAM_EnumFormat, + SPA_FORMAT_mediaType, SPA_POD_Id(SPA_MEDIA_TYPE_application), + SPA_FORMAT_mediaSubtype, SPA_POD_Id(SPA_MEDIA_SUBTYPE_control)); + break; + + case SPA_PARAM_Format: + if (!port->have_format) + return -EIO; + if (result.index > 0) + return 0; + + param = spa_pod_builder_add_object(&b, + SPA_TYPE_OBJECT_Format, SPA_PARAM_Format, + SPA_FORMAT_mediaType, SPA_POD_Id(SPA_MEDIA_TYPE_application), + SPA_FORMAT_mediaSubtype, SPA_POD_Id(SPA_MEDIA_SUBTYPE_control)); + 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, 1, MAX_BUFFERS), + SPA_PARAM_BUFFERS_blocks, SPA_POD_Int(1), + SPA_PARAM_BUFFERS_size, SPA_POD_CHOICE_RANGE_Int( + 4096, 4096, INT32_MAX), + SPA_PARAM_BUFFERS_stride, SPA_POD_Int(1)); + 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; + + case SPA_PARAM_IO: + switch (result.index) { + case 0: + param = spa_pod_builder_add_object(&b, + SPA_TYPE_OBJECT_ParamIO, id, + SPA_PARAM_IO_id, SPA_POD_Id(SPA_IO_Buffers), + SPA_PARAM_IO_size, SPA_POD_Int(sizeof(struct spa_io_buffers))); + break; + default: + return 0; + } + break; + + case SPA_PARAM_Latency: + switch (result.index) { + case 0: + param = spa_latency_build(&b, id, &port->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 port_set_format(struct impl *this, struct port *port, + uint32_t flags, + const struct spa_pod *format) +{ + int err; + + if (format == NULL) { + if (!port->have_format) + return 0; + + 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_application || + info.media_subtype != SPA_MEDIA_SUBTYPE_control) + return -EINVAL; + + port->current_format = info; + port->have_format = true; + } + + port->info.change_mask |= SPA_PORT_CHANGE_MASK_RATE; + port->info.rate = SPA_FRACTION(1, 1); + port->info.change_mask |= SPA_PORT_CHANGE_MASK_PARAMS; + if (port->have_format) { + port->params[IDX_Format] = SPA_PARAM_INFO(SPA_PARAM_Format, SPA_PARAM_INFO_READWRITE); + port->params[IDX_Buffers] = SPA_PARAM_INFO(SPA_PARAM_Buffers, SPA_PARAM_INFO_READ); + } else { + port->params[IDX_Format] = SPA_PARAM_INFO(SPA_PARAM_Format, SPA_PARAM_INFO_WRITE); + port->params[IDX_Buffers] = 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(this, direction, port_id), -EINVAL); + + port = GET_PORT(this, direction, port_id); + + switch (id) { + case SPA_PARAM_Format: + res = port_set_format(this, port, flags, param); + break; + case SPA_PARAM_Latency: + res = 0; + 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 = GET_PORT(this, direction, port_id); + + spa_log_debug(this->log, "%p: use buffers %d", this, 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].data == NULL) { + spa_log_error(this->log, "%p: need mapped memory", this); + return -EINVAL; + } + } + port->n_buffers = n_buffers; + + reset_buffers(port); + + 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 = GET_PORT(this, direction, port_id); + + 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) +{ + struct impl *this = object; + struct port *port; + + spa_return_val_if_fail(this != NULL, -EINVAL); + + spa_return_val_if_fail(CHECK_PORT(this, SPA_DIRECTION_OUTPUT, port_id), -EINVAL); + + port = GET_PORT(this, SPA_DIRECTION_OUTPUT, port_id); + + 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; + int status = SPA_STATUS_OK; + + spa_return_val_if_fail(this != NULL, -EINVAL); + + if (!this->started) + return SPA_STATUS_OK; + + if (this->following) { + if (this->position) { + this->current_time = this->position->clock.nsec; + } else { + struct timespec now; + spa_system_clock_gettime(this->data_system, CLOCK_MONOTONIC, &now); + this->current_time = SPA_TIMESPEC_TO_NSEC(&now); + } + } + + update_position(this); + + if (this->following) + status |= process_output(this); + + status |= process_input(this); + + return status; +} + +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 int impl_get_interface(struct spa_handle *handle, const char *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 (spa_streq(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; + + do_release(this); + + free(this->chr_path); + if (this->timerfd > 0) + spa_system_close(this->data_system, this->timerfd); + if (this->conn) + dbus_connection_unref(this->conn); + if (this->dbus_connection) + spa_dbus_connection_destroy(this->dbus_connection); + + spa_zero(*this); + + 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; + int res = 0; + size_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; + + this->log = spa_support_find(support, n_support, SPA_TYPE_INTERFACE_Log); + this->data_loop = spa_support_find(support, n_support, SPA_TYPE_INTERFACE_DataLoop); + this->data_system = spa_support_find(support, n_support, SPA_TYPE_INTERFACE_DataSystem); + this->dbus = spa_support_find(support, n_support, SPA_TYPE_INTERFACE_DBus); + + if (this->log == NULL) + return -EINVAL; + + spa_log_topic_init(this->log, &log_topic); + + 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; + } + if (this->dbus == NULL) { + spa_log_error(this->log, "dbus is needed"); + return -EINVAL; + } + + if (info) { + const char *str; + + if ((str = spa_dict_lookup(info, SPA_KEY_API_BLUEZ5_PATH)) != NULL) + this->chr_path = strdup(str); + } + + if (this->chr_path == NULL) { + spa_log_error(this->log, "missing MIDI service characteristic path"); + res = -EINVAL; + goto fail; + } + + this->dbus_connection = spa_dbus_get_connection(this->dbus, SPA_DBUS_TYPE_SYSTEM); + if (this->dbus_connection == NULL) { + spa_log_error(this->log, "no dbus connection"); + res = -EIO; + goto fail; + } + + this->conn = spa_dbus_connection_get(this->dbus_connection); + if (this->conn == NULL) { + spa_log_error(this->log, "failed to get dbus connection"); + res = -EIO; + goto fail; + } + + /* + * XXX: We should handle spa_dbus reconnecting, but we don't, so ref + * XXX: the handle so that we can keep it if spa_dbus unrefs it. + */ + dbus_connection_ref(this->conn); + + 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.max_input_ports = 1; + this->info.max_output_ports = 1; + this->info.flags = SPA_NODE_FLAG_RT; + this->params[IDX_PropInfo] = SPA_PARAM_INFO(SPA_PARAM_PropInfo, SPA_PARAM_INFO_READ); + this->params[IDX_Props] = SPA_PARAM_INFO(SPA_PARAM_Props, SPA_PARAM_INFO_READWRITE); + this->params[IDX_NODE_IO] = SPA_PARAM_INFO(SPA_PARAM_IO, SPA_PARAM_INFO_READ); + this->info.params = this->params; + this->info.n_params = N_NODE_PARAMS; + + /* set the port info */ + for (i = 0; i < N_PORTS; ++i) { + struct port *port = &this->ports[i]; + static const struct spa_dict_item in_port_items[] = { + SPA_DICT_ITEM_INIT(SPA_KEY_FORMAT_DSP, "8 bit raw midi"), + SPA_DICT_ITEM_INIT(SPA_KEY_PORT_NAME, "in"), + SPA_DICT_ITEM_INIT(SPA_KEY_PORT_ALIAS, "in"), + }; + static const struct spa_dict_item out_port_items[] = { + SPA_DICT_ITEM_INIT(SPA_KEY_FORMAT_DSP, "8 bit raw midi"), + SPA_DICT_ITEM_INIT(SPA_KEY_PORT_NAME, "out"), + SPA_DICT_ITEM_INIT(SPA_KEY_PORT_ALIAS, "out"), + }; + static const struct spa_dict in_port_props = SPA_DICT_INIT_ARRAY(in_port_items); + static const struct spa_dict out_port_props = SPA_DICT_INIT_ARRAY(out_port_items); + + spa_zero(*port); + + port->impl = this; + + port->id = 0; + port->direction = (i == PORT_OUT) ? SPA_DIRECTION_OUTPUT : + SPA_DIRECTION_INPUT; + + port->info_all = SPA_PORT_CHANGE_MASK_FLAGS | + SPA_PORT_CHANGE_MASK_PROPS | + 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; + port->params[IDX_EnumFormat] = SPA_PARAM_INFO(SPA_PARAM_EnumFormat, SPA_PARAM_INFO_READ); + port->params[IDX_Meta] = SPA_PARAM_INFO(SPA_PARAM_Meta, SPA_PARAM_INFO_READ); + port->params[IDX_IO] = SPA_PARAM_INFO(SPA_PARAM_IO, SPA_PARAM_INFO_READ); + port->params[IDX_Format] = SPA_PARAM_INFO(SPA_PARAM_Format, SPA_PARAM_INFO_WRITE); + port->params[IDX_Buffers] = SPA_PARAM_INFO(SPA_PARAM_Buffers, 0); + port->params[IDX_Latency] = SPA_PARAM_INFO(SPA_PARAM_Latency, SPA_PARAM_INFO_READWRITE); + port->info.params = port->params; + port->info.n_params = N_PORT_PARAMS; + port->info.props = (i == PORT_OUT) ? &out_port_props : &in_port_props; + + port->latency = SPA_LATENCY_INFO(port->direction); + port->latency.min_quantum = 1.0f; + port->latency.max_quantum = 1.0f; + + /* Init the buffer lists */ + spa_list_init(&port->ready); + spa_list_init(&port->free); + } + + this->duration = 1024; + this->rate = 48000; + + set_latency(this, false); + + this->timerfd = spa_system_timerfd_create(this->data_system, + CLOCK_MONOTONIC, SPA_FD_CLOEXEC | SPA_FD_NONBLOCK); + + return 0; + +fail: + res = (res < 0) ? res : ((errno > 0) ? -errno : -EIO); + impl_clear(handle); + return res; +} + +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, "Pauli Virtanen " }, + { SPA_KEY_FACTORY_DESCRIPTION, "Bluez5 MIDI connection" }, +}; + +static const struct spa_dict info = SPA_DICT_INIT_ARRAY(info_items); + +const struct spa_handle_factory spa_bluez5_midi_node_factory = { + SPA_VERSION_HANDLE_FACTORY, + SPA_NAME_API_BLUEZ5_MIDI_NODE, + &info, + impl_get_size, + impl_init, + impl_enum_interface_info, +}; diff --git a/spa/plugins/bluez5/plugin.c b/spa/plugins/bluez5/plugin.c index dbbce9bf1..7b7f8627f 100644 --- a/spa/plugins/bluez5/plugin.c +++ b/spa/plugins/bluez5/plugin.c @@ -36,6 +36,7 @@ extern const struct spa_handle_factory spa_sco_source_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_bluez5_midi_enum_factory; +extern const struct spa_handle_factory spa_bluez5_midi_node_factory; SPA_EXPORT int spa_handle_factory_enum(const struct spa_handle_factory **factory, uint32_t *index) @@ -71,6 +72,9 @@ int spa_handle_factory_enum(const struct spa_handle_factory **factory, uint32_t case 8: *factory = &spa_bluez5_midi_enum_factory; break; + case 9: + *factory = &spa_bluez5_midi_node_factory; + break; default: return 0; }