- pthread_mutex_lock(&consumer_data.lock);
- pthread_mutex_lock(&stream->lock);
-
- destroy_monitor(stream, ht);
-
- /* Update refcount of channel and see if we need to destroy it. */
- if (!uatomic_sub_return(&stream->chan->refcount, 1)
- && !uatomic_read(&stream->chan->nb_init_stream_left)) {
- free_chan = stream->chan;
+ /*
+ * This means that the stream was successfully removed from the streams
+ * list of the channel and sent to the right thread managing this
+ * stream thus being globally visible.
+ */
+ if (stream->globally_visible) {
+ pthread_mutex_lock(&consumer_data.lock);
+ pthread_mutex_lock(&stream->chan->lock);
+ pthread_mutex_lock(&stream->lock);
+ /* Remove every reference of the stream in the consumer. */
+ consumer_stream_delete(stream, ht);
+
+ destroy_close_stream(stream);
+
+ /* Update channel's refcount of the stream. */
+ free_chan = unref_channel(stream);
+
+ /* Indicates that the consumer data state MUST be updated after this. */
+ consumer_data.need_update = 1;
+
+ pthread_mutex_unlock(&stream->lock);
+ pthread_mutex_unlock(&stream->chan->lock);
+ pthread_mutex_unlock(&consumer_data.lock);
+ } else {
+ /*
+ * If the stream is not visible globally, this needs to be done
+ * outside of the consumer data lock section.
+ */
+ free_chan = unref_channel(stream);