bluez5: change sink/source run state follow transport state

Allow asynchronous changes in transport state in the sinks/sources.

Also allow transport acquire to be actually synchronous, in this case it
must set transport state during acquire call.

Separate driver start/stop from transport start/stop.
This commit is contained in:
Pauli Virtanen 2023-03-06 22:40:33 +02:00 committed by P V
parent 60718c4b4f
commit 1d5c693d33
5 changed files with 439 additions and 145 deletions

View file

@ -539,9 +539,10 @@ int spa_bt_sco_io_write(struct spa_bt_sco_io *io, uint8_t *data, int size);
#define SPA_BT_VOLUME_A2DP_MAX 127
enum spa_bt_transport_state {
SPA_BT_TRANSPORT_STATE_IDLE,
SPA_BT_TRANSPORT_STATE_PENDING,
SPA_BT_TRANSPORT_STATE_ACTIVE,
SPA_BT_TRANSPORT_STATE_ERROR = -1,
SPA_BT_TRANSPORT_STATE_IDLE = 0,
SPA_BT_TRANSPORT_STATE_PENDING = 1,
SPA_BT_TRANSPORT_STATE_ACTIVE = 2,
};
struct spa_bt_transport_events {

View file

@ -116,6 +116,8 @@ struct impl {
struct port port;
unsigned int started:1;
unsigned int start_ready:1;
unsigned int transport_started:1;
unsigned int following:1;
unsigned int is_output:1;
unsigned int flush_pending:1;
@ -617,6 +619,8 @@ static int flush_data(struct impl *this, uint64_t now_time)
struct port *port = &this->port;
int unused_buffer;
spa_assert(this->transport_started);
if (this->transport == NULL || !this->flush_source.loop || !this->flush_timer_source.loop) {
/* I/O in error state */
return -EIO;
@ -907,25 +911,22 @@ static void media_on_timeout(struct spa_source *source)
set_timeout(this, this->next_time);
}
static int do_start(struct impl *this)
static int transport_start(struct impl *this)
{
int res, val, size;
int val, size;
struct port *port;
socklen_t len;
uint8_t *conf;
uint32_t flags;
if (this->started)
if (this->transport_started)
return 0;
if (!this->start_ready)
return -EIO;
spa_return_val_if_fail(this->transport, -EIO);
this->following = is_following(this);
spa_log_debug(this->log, "%p: start following:%d", this, this->following);
if ((res = spa_bt_transport_acquire(this->transport, false)) < 0)
return res;
spa_log_debug(this->log, "%p: start transport", this);
port = &this->port;
@ -984,13 +985,6 @@ static int do_start(struct impl *this)
reset_buffer(this);
this->source.data = this;
this->source.fd = this->timerfd;
this->source.func = media_on_timeout;
this->source.mask = SPA_IO_IN;
this->source.rmask = 0;
spa_loop_add_source(this->data_loop, &this->source);
this->flush_timer_source.data = this;
this->flush_timer_source.fd = this->flush_timerfd;
this->flush_timer_source.func = media_on_flush_timeout;
@ -1007,7 +1001,40 @@ static int do_start(struct impl *this)
this->flush_pending = false;
this->transport_started = true;
return 0;
}
static int do_start(struct impl *this)
{
int res;
if (this->started)
return 0;
spa_return_val_if_fail(this->transport, -EIO);
this->following = is_following(this);
spa_log_debug(this->log, "%p: start following:%d", this, this->following);
this->start_ready = true;
if ((res = spa_bt_transport_acquire(this->transport, false)) < 0) {
this->start_ready = false;
return res;
}
this->source.data = this;
this->source.fd = this->timerfd;
this->source.func = media_on_timeout;
this->source.mask = SPA_IO_IN;
this->source.rmask = 0;
spa_loop_add_source(this->data_loop, &this->source);
set_timers(this);
this->started = true;
return 0;
@ -1031,6 +1058,21 @@ static int do_remove_source(struct spa_loop *loop,
ts.it_interval.tv_nsec = 0;
spa_system_timerfd_settime(this->data_system, this->timerfd, 0, &ts, NULL);
return 0;
}
static int do_remove_transport_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;
this->transport_started = false;
if (this->flush_source.loop)
spa_loop_remove_source(this->data_loop, &this->flush_source);
@ -1045,6 +1087,20 @@ static int do_remove_source(struct spa_loop *loop,
return 0;
}
static void transport_stop(struct impl *this)
{
if (!this->transport_started)
return;
spa_log_trace(this->log, "%p: stop transport", this);
spa_loop_invoke(this->data_loop, do_remove_transport_source, 0, NULL, 0, true, this);
if (this->codec_data)
this->codec->deinit(this->codec_data);
this->codec_data = NULL;
}
static int do_stop(struct impl *this)
{
int res = 0;
@ -1052,18 +1108,18 @@ static int do_stop(struct impl *this)
if (!this->started)
return 0;
spa_log_trace(this->log, "%p: stop", this);
spa_log_debug(this->log, "%p: stop", this);
this->start_ready = false;
spa_loop_invoke(this->data_loop, do_remove_source, 0, NULL, 0, true, this);
this->started = false;
transport_stop(this);
if (this->transport)
res = spa_bt_transport_release(this->transport);
if (this->codec_data)
this->codec->deinit(this->codec_data);
this->codec_data = NULL;
this->started = false;
return res;
}
@ -1506,6 +1562,9 @@ static int impl_node_process(void *object)
return SPA_STATUS_HAVE_DATA;
}
if (!this->started || !this->transport_started)
return SPA_STATUS_OK;
if (io->status == SPA_STATUS_HAVE_DATA && io->buffer_id < port->n_buffers) {
struct buffer *b = &port->buffers[io->buffer_id];
@ -1598,26 +1657,32 @@ static void transport_state_changed(void *data,
enum spa_bt_transport_state state)
{
struct impl *this = data;
bool was_started = this->transport_started;
spa_log_debug(this->log, "%p: transport %p state %d->%d", this, this->transport, old, state);
if (state < SPA_BT_TRANSPORT_STATE_ACTIVE && old == SPA_BT_TRANSPORT_STATE_ACTIVE &&
this->started) {
uint8_t buffer[1024];
struct spa_pod_builder b = { 0 };
spa_log_debug(this->log, "%p: transport %p becomes inactive: stop and indicate error",
this, this->transport);
if (state == SPA_BT_TRANSPORT_STATE_ACTIVE)
transport_start(this);
else
transport_stop(this);
if (state < SPA_BT_TRANSPORT_STATE_ACTIVE && was_started) {
/*
* If establishing connection fails due to remote end not activating
* the transport, we won't get a write error, but instead see a transport
* state change.
*
* Stop and emit a node error, to let upper levels handle it.
* Emit a node error, to let upper levels handle it.
*/
do_stop(this);
spa_log_debug(this->log, "%p: transport %p becomes inactive: stop and indicate error",
this, this->transport);
state = SPA_BT_TRANSPORT_STATE_ERROR;
}
if (state == SPA_BT_TRANSPORT_STATE_ERROR) {
uint8_t buffer[1024];
struct spa_pod_builder b = { 0 };
spa_pod_builder_init(&b, buffer, sizeof(buffer));
spa_node_emit_event(&this->hooks,

View file

@ -114,7 +114,8 @@ struct impl {
struct port port;
unsigned int started:1;
unsigned int transport_acquired:1;
unsigned int start_ready:1;
unsigned int transport_started:1;
unsigned int following:1;
unsigned int matching:1;
unsigned int resampling:1;
@ -258,7 +259,8 @@ static int do_reassign_follower(struct spa_loop *loop,
struct port *port = &this->port;
set_timers(this);
spa_bt_decode_buffer_recover(&port->buffer);
if (this->transport_started)
spa_bt_decode_buffer_recover(&port->buffer);
return 0;
}
@ -561,6 +563,9 @@ static int setup_matching(struct impl *this)
{
struct port *port = &this->port;
if (!this->transport_started)
port->buffer.corr = 1.0;
if (this->position && port->rate_match) {
port->rate_match->rate = 1 / port->buffer.corr;
@ -643,15 +648,15 @@ static int transport_start(struct impl *this)
struct port *port = &this->port;
uint32_t flags;
if (this->transport_acquired)
if (this->transport_started)
return 0;
if (!this->start_ready)
return -EIO;
spa_log_debug(this->log, "%p: transport %p acquire", this,
this->transport);
if ((res = spa_bt_transport_acquire(this->transport, false)) < 0)
return res;
spa_return_val_if_fail(this->transport != NULL, -EIO);
this->transport_acquired = true;
spa_log_debug(this->log, "%p: start transport state:%d",
this, this->transport->state);
flags = this->is_duplex ? 0 : MEDIA_CODEC_FLAG_SINK;
@ -724,25 +729,16 @@ static int transport_start(struct impl *this)
set_duplex_timeout(this, this->duplex_timeout);
}
this->timer_source.data = this;
this->timer_source.fd = this->timerfd;
this->timer_source.func = media_on_timeout;
this->timer_source.mask = SPA_IO_IN;
this->timer_source.rmask = 0;
spa_loop_add_source(this->data_loop, &this->timer_source);
this->sample_count = 0;
setup_matching(this);
set_timers(this);
this->transport_started = true;
return 0;
}
static int do_start(struct impl *this)
{
int res = 0;
int res;
if (this->started)
return 0;
@ -751,16 +747,31 @@ static int do_start(struct impl *this)
this->following = is_following(this);
spa_log_debug(this->log, "%p: start state:%d following:%d",
this, this->transport->state, this->following);
this->start_ready = true;
if (this->transport->state >= SPA_BT_TRANSPORT_STATE_PENDING ||
this->is_duplex || this->codec->bap)
res = transport_start(this);
spa_log_debug(this->log, "%p: start following:%d", this, this->following);
spa_log_debug(this->log, "%p: transport %p acquire", this,
this->transport);
if ((res = spa_bt_transport_acquire(this->transport, false)) < 0) {
this->start_ready = false;
return res;
}
this->timer_source.data = this;
this->timer_source.fd = this->timerfd;
this->timer_source.func = media_on_timeout;
this->timer_source.mask = SPA_IO_IN;
this->timer_source.rmask = 0;
spa_loop_add_source(this->data_loop, &this->timer_source);
setup_matching(this);
set_timers(this);
this->started = true;
return res;
return 0;
}
static int do_remove_source(struct spa_loop *loop,
@ -775,11 +786,6 @@ static int do_remove_source(struct spa_loop *loop,
spa_log_debug(this->log, "%p: remove source", this);
set_duplex_timeout(this, 0);
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;
@ -791,34 +797,48 @@ static int do_remove_source(struct spa_loop *loop,
return 0;
}
static int transport_stop(struct impl *this)
static int do_remove_transport_source(struct spa_loop *loop,
bool async,
uint32_t seq,
const void *data,
size_t size,
void *user_data)
{
struct impl *this = user_data;
spa_log_debug(this->log, "%p: remove transport source", this);
this->transport_started = false;
set_duplex_timeout(this, 0);
if (this->source.loop)
spa_loop_remove_source(this->data_loop, &this->source);
return 0;
}
static void transport_stop(struct impl *this)
{
struct port *port = &this->port;
int res;
if (!this->transport_started)
return;
spa_log_debug(this->log, "%p: transport stop", this);
spa_loop_invoke(this->data_loop, do_remove_source, 0, NULL, 0, true, this);
spa_loop_invoke(this->data_loop, do_remove_transport_source, 0, NULL, 0, true, this);
if (this->fd >= 0) {
close(this->fd);
this->fd = -1;
}
if (this->transport && this->transport_acquired)
res = spa_bt_transport_release(this->transport);
else
res = 0;
this->transport_acquired = false;
if (this->codec_data)
this->codec->deinit(this->codec_data);
this->codec_data = NULL;
spa_bt_decode_buffer_clear(&port->buffer);
return res;
}
static int do_stop(struct impl *this)
@ -830,7 +850,16 @@ static int do_stop(struct impl *this)
spa_log_debug(this->log, "%p: stop", this);
res = transport_stop(this);
this->start_ready = false;
spa_loop_invoke(this->data_loop, do_remove_source, 0, NULL, 0, true, this);
transport_stop(this);
if (this->transport)
res = spa_bt_transport_release(this->transport);
else
res = 0;
this->started = false;
@ -1450,7 +1479,7 @@ static int produce_buffer(struct impl *this)
}
/* Handle buffering */
if (this->started)
if (this->transport_started)
process_buffering(this);
/* Return if there are no buffers ready to be processed */
@ -1482,6 +1511,9 @@ static int impl_node_process(void *object)
if ((io = port->io) == NULL)
return -EIO;
if (!this->started || !this->transport_started)
return SPA_STATUS_OK;
spa_log_trace(this->log, "%p status:%d", this, io->status);
/* Return if we already have a buffer */
@ -1520,6 +1552,30 @@ static const struct spa_node_methods impl_node = {
.process = impl_node_process,
};
static void transport_state_changed(void *data,
enum spa_bt_transport_state old,
enum spa_bt_transport_state state)
{
struct impl *this = data;
spa_log_debug(this->log, "%p: transport %p state %d->%d", this, this->transport, old, state);
if (state == SPA_BT_TRANSPORT_STATE_ACTIVE)
transport_start(this);
else
transport_stop(this);
if (state == SPA_BT_TRANSPORT_STATE_ERROR) {
uint8_t buffer[1024];
struct spa_pod_builder b = { 0 };
spa_pod_builder_init(&b, buffer, sizeof(buffer));
spa_node_emit_event(&this->hooks,
spa_pod_builder_add_object(&b,
SPA_TYPE_EVENT_Node, SPA_NODE_EVENT_Error));
}
}
static void transport_delay_changed(void *data)
{
struct impl *this = data;
@ -1537,7 +1593,6 @@ static int do_transport_destroy(struct spa_loop *loop,
{
struct impl *this = user_data;
this->transport = NULL;
this->transport_acquired = false;
return 0;
}
@ -1551,6 +1606,7 @@ static void transport_destroy(void *data)
static const struct spa_bt_transport_events transport_events = {
SPA_VERSION_BT_TRANSPORT_EVENTS,
.delay_changed = transport_delay_changed,
.state_changed = transport_state_changed,
.destroy = transport_destroy,
};

View file

@ -116,6 +116,8 @@ struct impl {
/* Flags */
unsigned int started:1;
unsigned int start_ready:1;
unsigned int transport_started:1;
unsigned int following:1;
unsigned int flush_pending:1;
@ -366,6 +368,8 @@ static int flush_data(struct impl *this)
int processed = 0;
int written;
spa_assert(this->transport_started);
if (this->transport == NULL || this->transport->sco_io == NULL || !this->flush_timer_source.loop)
return -EIO;
@ -638,28 +642,22 @@ static int lcm(int a, int b) {
return (a*b)/gcd(a,b);
}
static int do_start(struct impl *this)
static int transport_start(struct impl *this)
{
bool do_accept;
int res;
/* Don't do anything if the node has already started */
if (this->started)
if (this->transport_started)
return 0;
if (!this->start_ready)
return -EIO;
/* Make sure the transport is valid */
spa_return_val_if_fail(this->transport != NULL, -EIO);
this->following = is_following(this);
spa_log_debug(this->log, "%p: start following:%d", this, this->following);
/* Do accept if Gateway; otherwise do connect for Head Unit */
do_accept = this->transport->profile & SPA_BT_PROFILE_HEADSET_AUDIO_GATEWAY;
/* acquire the socket fd (false -> connect | true -> accept) */
if ((res = spa_bt_transport_acquire(this->transport, do_accept)) < 0)
return res;
spa_log_debug(this->log, "%p: start transport", this);
/* Init mSBC if needed */
if (this->transport->codec == HFP_AUDIO_CODEC_MSBC) {
@ -688,14 +686,6 @@ static int do_start(struct impl *this)
if ((res = spa_bt_transport_ensure_sco_io(this->transport, this->data_loop)) < 0)
goto fail;
/* 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);
this->flush_timer_source.data = this;
this->flush_timer_source.fd = this->flush_timerfd;
this->flush_timer_source.func = sco_on_flush_timeout;
@ -705,20 +695,58 @@ static int do_start(struct impl *this)
/* start processing */
this->flush_pending = false;
set_timers(this);
/* Set the started flag */
this->started = true;
this->transport_started = true;
return 0;
fail:
free(this->buffer);
this->buffer = NULL;
spa_bt_transport_release(this->transport);
return res;
}
static int do_start(struct impl *this)
{
bool do_accept;
int res;
if (this->started)
return 0;
spa_return_val_if_fail(this->transport, -EIO);
this->following = is_following(this);
this->start_ready = true;
spa_log_debug(this->log, "%p: start following:%d", this, this->following);
/* Do accept if Gateway; otherwise do connect for Head Unit */
do_accept = this->transport->profile & SPA_BT_PROFILE_HEADSET_AUDIO_GATEWAY;
/* acquire the socket fd (false -> connect | true -> accept) */
if ((res = spa_bt_transport_acquire(this->transport, do_accept)) < 0) {
this->start_ready = false;
return res;
}
/* 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);
set_timers(this);
this->started = true;
return 0;
}
/* Drop any buffered data remaining in the port */
static void drop_port_output(struct impl *this)
{
@ -747,12 +775,26 @@ static int do_remove_source(struct spa_loop *loop,
void *user_data)
{
struct impl *this = user_data;
struct itimerspec ts;
set_timeout(this, 0);
if (this->source.loop)
spa_loop_remove_source(this->data_loop, &this->source);
return 0;
}
static int do_remove_transport_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;
this->transport_started = false;
if (this->flush_timer_source.loop)
spa_loop_remove_source(this->data_loop, &this->flush_timer_source);
ts.it_value.tv_sec = 0;
@ -767,29 +809,43 @@ static int do_remove_source(struct spa_loop *loop,
return 0;
}
static int do_stop(struct impl *this)
static void transport_stop(struct impl *this)
{
int res = 0;
if (!this->transport_started)
return;
if (!this->started)
return 0;
spa_log_trace(this->log, "sco-sink %p: transport stop", this);
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;
spa_loop_invoke(this->data_loop, do_remove_transport_source, 0, NULL, 0, true, this);
if (this->buffer) {
free(this->buffer);
this->buffer = NULL;
this->buffer_head = this->buffer_next = this->buffer;
}
}
if (this->transport) {
/* Release the transport; it is responsible for closing the fd */
static int do_stop(struct impl *this)
{
int res;
if (!this->started)
return 0;
spa_log_debug(this->log, "%p: stop", this);
this->start_ready = false;
spa_loop_invoke(this->data_loop, do_remove_source, 0, NULL, 0, true, this);
transport_stop(this);
if (this->transport)
res = spa_bt_transport_release(this->transport);
}
else
res = 0;
this->started = false;
return res;
}
@ -1241,6 +1297,9 @@ static int impl_node_process(void *object)
return SPA_STATUS_HAVE_DATA;
}
if (!this->started || !this->transport_started)
return SPA_STATUS_OK;
if (io->status == SPA_STATUS_HAVE_DATA && io->buffer_id < port->n_buffers) {
struct buffer *b = &port->buffers[io->buffer_id];
@ -1301,6 +1360,30 @@ static const struct spa_node_methods impl_node = {
.process = impl_node_process,
};
static void transport_state_changed(void *data,
enum spa_bt_transport_state old,
enum spa_bt_transport_state state)
{
struct impl *this = data;
spa_log_debug(this->log, "%p: transport %p state %d->%d", this, this->transport, old, state);
if (state == SPA_BT_TRANSPORT_STATE_ACTIVE)
transport_start(this);
else if (state < SPA_BT_TRANSPORT_STATE_ACTIVE)
transport_stop(this);
if (state == SPA_BT_TRANSPORT_STATE_ERROR) {
uint8_t buffer[1024];
struct spa_pod_builder b = { 0 };
spa_pod_builder_init(&b, buffer, sizeof(buffer));
spa_node_emit_event(&this->hooks,
spa_pod_builder_add_object(&b,
SPA_TYPE_EVENT_Node, SPA_NODE_EVENT_Error));
}
}
static int do_transport_destroy(struct spa_loop *loop,
bool async,
uint32_t seq,
@ -1322,6 +1405,7 @@ static void transport_destroy(void *data)
static const struct spa_bt_transport_events transport_events = {
SPA_VERSION_BT_TRANSPORT_EVENTS,
.state_changed = transport_state_changed,
.destroy = transport_destroy,
};

View file

@ -113,6 +113,8 @@ struct impl {
struct port port;
unsigned int started:1;
unsigned int start_ready:1;
unsigned int transport_started:1;
unsigned int following:1;
unsigned int matching:1;
unsigned int resampling:1;
@ -231,7 +233,8 @@ static int do_reassign_follower(struct spa_loop *loop,
struct port *port = &this->port;
set_timers(this);
spa_bt_decode_buffer_recover(&port->buffer);
if (this->transport_started)
spa_bt_decode_buffer_recover(&port->buffer);
return 0;
}
@ -496,6 +499,10 @@ static int sco_source_cb(void *userdata, uint8_t *read_data, int size_read)
uint32_t decoded;
uint64_t dt;
/* Drop data when not started */
if (!this->started)
return 0;
if (this->transport == NULL) {
spa_log_debug(this->log, "no transport, stop reading");
goto stop;
@ -557,6 +564,9 @@ static int setup_matching(struct impl *this)
{
struct port *port = &this->port;
if (!this->transport_started)
port->buffer.corr = 1.0;
if (this->position && port->rate_match) {
port->rate_match->rate = 1 / port->buffer.corr;
@ -645,31 +655,22 @@ static int do_add_source(struct spa_loop *loop,
return 0;
}
static int do_start(struct impl *this)
static int transport_start(struct impl *this)
{
struct port *port = &this->port;
bool do_accept;
int res;
/* Don't do anything if the node has already started */
if (this->started)
if (this->transport_started)
return 0;
if (!this->start_ready)
return -EIO;
this->following = is_following(this);
spa_log_debug(this->log, "%p: start following:%d",
this, this->following);
spa_log_debug(this->log, "%p: start transport", this);
/* 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 socket fd (false -> connect | true -> accept) */
if ((res = spa_bt_transport_acquire(this->transport, do_accept)) < 0)
return res;
/* Reset the buffers and sample count */
reset_buffers(port);
@ -700,6 +701,40 @@ static int do_start(struct impl *this)
goto fail;
spa_loop_invoke(this->data_loop, do_add_source, 0, NULL, 0, true, this);
/* Set the started flag */
this->transport_started = true;
return 0;
fail:
return res;
}
static int do_start(struct impl *this)
{
bool do_accept;
int res;
if (this->started)
return 0;
spa_return_val_if_fail(this->transport, -EIO);
this->following = is_following(this);
this->start_ready = true;
spa_log_debug(this->log, "%p: start following:%d", this, this->following);
/* Do accept if Gateway; otherwise do connect for Head Unit */
do_accept = this->transport->profile & SPA_BT_PROFILE_HEADSET_AUDIO_GATEWAY;
/* acquire the socket fd (false -> connect | true -> accept) */
if ((res = spa_bt_transport_acquire(this->transport, do_accept)) < 0) {
this->start_ready = false;
return res;
}
/* Start timer */
this->timer_source.data = this;
this->timer_source.fd = this->timerfd;
@ -709,16 +744,12 @@ static int do_start(struct impl *this)
spa_loop_add_source(this->data_loop, &this->timer_source);
setup_matching(this);
set_timers(this);
/* Set the started flag */
this->started = true;
return 0;
fail:
spa_bt_transport_release(this->transport);
return res;
}
static int do_remove_source(struct spa_loop *loop,
@ -731,9 +762,6 @@ static int do_remove_source(struct spa_loop *loop,
struct impl *this = user_data;
struct itimerspec ts;
if (this->transport && this->transport->sco_io)
spa_bt_sco_io_set_source_cb(this->transport->sco_io, NULL, NULL);
if (this->timer_source.loop)
spa_loop_remove_source(this->data_loop, &this->timer_source);
ts.it_value.tv_sec = 0;
@ -745,26 +773,58 @@ static int do_remove_source(struct spa_loop *loop,
return 0;
}
static int do_stop(struct impl *this)
static int do_remove_transport_source(struct spa_loop *loop,
bool async,
uint32_t seq,
const void *data,
size_t size,
void *user_data)
{
struct impl *this = user_data;
this->transport_started = false;
if (this->transport && this->transport->sco_io)
spa_bt_sco_io_set_source_cb(this->transport->sco_io, NULL, NULL);
return 0;
}
static void transport_stop(struct impl *this)
{
struct port *port = &this->port;
int res = 0;
if (!this->transport_started)
return;
spa_log_debug(this->log, "sco-source %p: transport stop", this);
spa_loop_invoke(this->data_loop, do_remove_transport_source, 0, NULL, 0, true, this);
spa_bt_decode_buffer_clear(&port->buffer);
}
static int do_stop(struct impl *this)
{
int res;
if (!this->started)
return 0;
spa_log_debug(this->log, "sco-source %p: stop", this);
spa_log_debug(this->log, "%p: stop", this);
this->start_ready = false;
spa_loop_invoke(this->data_loop, do_remove_source, 0, NULL, 0, true, this);
this->started = false;
transport_stop(this);
if (this->transport) {
/* Release the transport; it is responsible for closing the fd */
if (this->transport)
res = spa_bt_transport_release(this->transport);
}
else
res = 0;
spa_bt_decode_buffer_clear(&port->buffer);
this->started = false;
return res;
}
@ -1322,7 +1382,7 @@ static int produce_buffer(struct impl *this)
}
/* Handle buffering */
if (this->started)
if (this->transport_started)
process_buffering(this);
/* Return if there are no buffers ready to be processed */
@ -1354,6 +1414,9 @@ static int impl_node_process(void *object)
if ((io = port->io) == NULL)
return -EIO;
if (!this->started || !this->transport_started)
return SPA_STATUS_OK;
spa_log_trace(this->log, "%p status:%d", this, io->status);
/* Return if we already have a buffer */
@ -1392,6 +1455,30 @@ static const struct spa_node_methods impl_node = {
.process = impl_node_process,
};
static void transport_state_changed(void *data,
enum spa_bt_transport_state old,
enum spa_bt_transport_state state)
{
struct impl *this = data;
spa_log_debug(this->log, "%p: transport %p state %d->%d", this, this->transport, old, state);
if (state == SPA_BT_TRANSPORT_STATE_ACTIVE)
transport_start(this);
else if (state < SPA_BT_TRANSPORT_STATE_ACTIVE)
transport_stop(this);
if (state == SPA_BT_TRANSPORT_STATE_ERROR) {
uint8_t buffer[1024];
struct spa_pod_builder b = { 0 };
spa_pod_builder_init(&b, buffer, sizeof(buffer));
spa_node_emit_event(&this->hooks,
spa_pod_builder_add_object(&b,
SPA_TYPE_EVENT_Node, SPA_NODE_EVENT_Error));
}
}
static int do_transport_destroy(struct spa_loop *loop,
bool async,
uint32_t seq,
@ -1413,6 +1500,7 @@ static void transport_destroy(void *data)
static const struct spa_bt_transport_events transport_events = {
SPA_VERSION_BT_TRANSPORT_EVENTS,
.state_changed = transport_state_changed,
.destroy = transport_destroy,
};