bluez5: guard against processing when errored or not started

Add some guards against doing processing when there has been an error or
the node is not started. Set error status to IO.  Continue driving on IO
errors.

spa/plugins/bluez5/media-sink.c

@@ -617,7 +617,7 @@ static int flush_data(struct impl *this, uint64_t now_time)
 	struct port *port = &this->port;
 	int unused_buffer;
 
-	if (!this->flush_source.loop) {
+	if (this->transport == NULL || !this->flush_source.loop || !this->flush_timer_source.loop) {
 		/* I/O in error state */
 		return -EIO;
 	}
@@ -814,6 +814,9 @@ static void media_on_flush_error(struct spa_source *source)
 		spa_log_warn(this->log, "%p: error %d", this, source->rmask);
 		if (this->flush_source.loop)
 			spa_loop_remove_source(this->data_loop, &this->flush_source);
+		enable_flush_timer(this, false);
+		if (this->flush_timer_source.loop)
+			spa_loop_remove_source(this->data_loop, &this->flush_timer_source);
 		return;
 	}
 }
@@ -853,9 +856,6 @@ static void media_on_timeout(struct spa_source *source)
 	uint64_t prev_time, now_time;
 	int res;
 
-	if (this->transport == NULL)
-		return;
-
 	if (this->started) {
 		if ((res = spa_system_timerfd_read(this->data_system, this->timerfd, &exp)) < 0) {
 			if (res != -EAGAIN)
@@ -882,7 +882,7 @@ static void media_on_timeout(struct spa_source *source)
 	this->next_time = now_time + duration * SPA_NSEC_PER_SEC / rate;
 
 	if (SPA_LIKELY(this->clock)) {
-		int64_t delay_nsec;
+		int64_t delay_nsec = 0;
 
 		this->clock->nsec = now_time;
 		this->clock->position += duration;
@@ -890,7 +890,8 @@ static void media_on_timeout(struct spa_source *source)
 		this->clock->rate_diff = 1.0f;
 		this->clock->next_nsec = this->next_time;
 
-		delay_nsec = spa_bt_transport_get_delay_nsec(this->transport);
+		if (this->transport)
+			delay_nsec = spa_bt_transport_get_delay_nsec(this->transport);
 
 		/* Negative delay doesn't work properly, so disallow it */
 		delay_nsec += SPA_CLAMP(this->props.latency_offset, -delay_nsec, INT64_MAX / 2);
@@ -1536,8 +1537,12 @@ static int impl_node_process(void *object)
 	this->process_time = this->current_time;
 
 	if (!spa_list_is_empty(&port->ready)) {
+		int res;
 		spa_log_trace(this->log, "%p: flush on process", this);
-		flush_data(this, this->current_time);
+		if ((res = flush_data(this, this->current_time)) < 0) {
+			io->status = res;
+			return SPA_STATUS_STOPPED;
+		}
 	}
 
 	return SPA_STATUS_HAVE_DATA;

spa/plugins/bluez5/media-source.c

@@ -1444,8 +1444,14 @@ static int produce_buffer(struct impl *this)
 		io->buffer_id = SPA_ID_INVALID;
 	}
 
+	if (!this->source.loop) {
+		io->status = -EIO;
+		return SPA_STATUS_STOPPED;
+	}
+
 	/* Handle buffering */
-	process_buffering(this);
+	if (this->started)
+		process_buffering(this);
 
 	/* Return if there are no buffers ready to be processed */
 	if (spa_list_is_empty(&port->ready))

spa/plugins/bluez5/sco-sink.c

@@ -360,9 +360,15 @@ static uint32_t get_queued_frames(struct impl *this)
 	return bytes / port->frame_size;
 }
 
-static void flush_data(struct impl *this)
+static int flush_data(struct impl *this)
 {
 	struct port *port = &this->port;
+	int processed = 0;
+	int written;
+
+	if (this->transport == NULL || this->transport->sco_io == NULL || !this->flush_timer_source.loop)
+		return -EIO;
+
 	const uint32_t min_in_size =
 		(this->transport->codec == HFP_AUDIO_CODEC_MSBC) ?
 		MSBC_DECODED_SIZE : this->transport->write_mtu;
@@ -372,18 +378,13 @@ static void flush_data(struct impl *this)
 	const uint32_t packet_samples = min_in_size / port->frame_size;
 	const uint64_t packet_time = (uint64_t)packet_samples * SPA_NSEC_PER_SEC
 		/ port->current_format.info.raw.rate;
-	int processed = 0;
-	int written;
-
-	if (this->transport == NULL || this->transport->sco_io == NULL)
-		return;
 
 	while (!spa_list_is_empty(&port->ready) && port->write_buffer_size < min_in_size) {
 		struct spa_data *datas;
 
 		/* get buffer */
 		if (!port->current_buffer) {
-			spa_return_if_fail(!spa_list_is_empty(&port->ready));
+			spa_return_val_if_fail(!spa_list_is_empty(&port->ready), -EIO);
 			port->current_buffer = spa_list_first(&port->ready, struct buffer, link);
 			port->ready_offset = 0;
 		}
@@ -418,14 +419,14 @@ static void flush_data(struct impl *this)
 
 	if (this->flush_pending) {
 		spa_log_trace(this->log, "%p: wait for flush timer", this);
-		return;
+		return 0;
 	}
 
 	if (port->write_buffer_size < min_in_size) {
 		/* wait for more data */
 		spa_log_trace(this->log, "%p: skip flush", this);
 		enable_flush_timer(this, false);
-		return;
+		return 0;
 	}
 
 	if (this->transport->codec == HFP_AUDIO_CODEC_MSBC) {
@@ -445,7 +446,7 @@ static void flush_data(struct impl *this)
 				this->buffer_next + 2, MSBC_ENCODED_SIZE - 3, &out_encoded);
 		if (processed < 0) {
 			spa_log_warn(this->log, "sbc_encode failed: %d", processed);
-			return;
+			return -EINVAL;
 		}
 		this->buffer_next += out_encoded + 3;
 		port->write_buffer_size = 0;
@@ -539,12 +540,13 @@ static void flush_data(struct impl *this)
 	}
 
 	enable_flush_timer(this, true);
-	return;
+	return 0;
 
 stop:
-	if (this->source.loop)
-		spa_loop_remove_source(this->data_loop, &this->source);
 	enable_flush_timer(this, false);
+	if (this->flush_timer_source.loop)
+		spa_loop_remove_source(this->data_loop, &this->flush_timer_source);
+	return -EIO;
 }
 
 static void sco_on_flush_timeout(struct spa_source *source)
@@ -582,9 +584,6 @@ static void sco_on_timeout(struct spa_source *source)
 	uint64_t prev_time, now_time;
 	int res;
 
-	if (this->transport == NULL)
-		return;
-
 	if (this->started) {
 		if ((res = spa_system_timerfd_read(this->data_system, this->timerfd, &exp)) < 0) {
 			if (res != -EAGAIN)
@@ -1272,8 +1271,12 @@ static int impl_node_process(void *object)
 	this->process_time = this->current_time;
 
 	if (!spa_list_is_empty(&port->ready)) {
+		int res;
 		spa_log_trace(this->log, "%p: flush on process", this);
-		flush_data(this);
+		if ((res = flush_data(this)) < 0) {
+			io->status = res;
+			return SPA_STATUS_STOPPED;
+		}
 	}
 
 	return SPA_STATUS_HAVE_DATA;

spa/plugins/bluez5/sco-source.c

@@ -116,6 +116,7 @@ struct impl {
 	unsigned int following:1;
 	unsigned int matching:1;
 	unsigned int resampling:1;
+	unsigned int io_error:1;
 
 	struct spa_source timer_source;
 	int timerfd;
@@ -548,6 +549,7 @@ static int sco_source_cb(void *userdata, uint8_t *read_data, int size_read)
 	return 0;
 
 stop:
+	this->io_error = true;
 	return 1;
 }
 
@@ -583,9 +585,6 @@ static void sco_on_timeout(struct spa_source *source)
 	uint64_t prev_time, now_time;
 	int res;
 
-	if (this->transport == NULL)
-		return;
-
 	if (this->started) {
 		if ((res = spa_system_timerfd_read(this->data_system, this->timerfd, &exp)) < 0) {
 			if (res != -EAGAIN)
@@ -694,6 +693,8 @@ static int do_start(struct impl *this)
 		this->msbc_buffer_pos = 0;
 	}
 
+	this->io_error = false;
+
 	/* Start socket i/o */
 	if ((res = spa_bt_transport_ensure_sco_io(this->transport, this->data_loop)) < 0)
 		goto fail;
@@ -1315,8 +1316,14 @@ static int produce_buffer(struct impl *this)
 		io->buffer_id = SPA_ID_INVALID;
 	}
 
+	if (this->io_error) {
+		io->status = -EIO;
+		return SPA_STATUS_STOPPED;
+	}
+
 	/* Handle buffering */
-	process_buffering(this);
+	if (this->started)
+		process_buffering(this);
 
 	/* Return if there are no buffers ready to be processed */
 	if (spa_list_is_empty(&port->ready))