diff --git a/spa/plugins/bluez5/media-sink.c b/spa/plugins/bluez5/media-sink.c
index 7a9531298..397270d36 100644
--- a/spa/plugins/bluez5/media-sink.c
+++ b/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;
diff --git a/spa/plugins/bluez5/media-source.c b/spa/plugins/bluez5/media-source.c
index e2d2ee1ae..94ca7ec36 100644
--- a/spa/plugins/bluez5/media-source.c
+++ b/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))
diff --git a/spa/plugins/bluez5/sco-sink.c b/spa/plugins/bluez5/sco-sink.c
index 6be5c486d..cddbc38bc 100644
--- a/spa/plugins/bluez5/sco-sink.c
+++ b/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;
diff --git a/spa/plugins/bluez5/sco-source.c b/spa/plugins/bluez5/sco-source.c
index 165ad20f6..d596211d7 100644
--- a/spa/plugins/bluez5/sco-source.c
+++ b/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))