+/*
+ * Send all stream of a channel to the right thread handling it.
+ *
+ * On error, return a negative value else 0 on success.
+ */
+static int send_streams_to_thread(struct lttng_consumer_channel *channel,
+ struct lttng_consumer_local_data *ctx)
+{
+ int ret = 0;
+ struct lttng_consumer_stream *stream, *stmp;
+
+ assert(channel);
+ assert(ctx);
+
+ /* Send streams to the corresponding thread. */
+ cds_list_for_each_entry_safe(stream, stmp, &channel->streams.head,
+ send_node) {
+ /* Sending the stream to the thread. */
+ ret = send_stream_to_thread(stream, ctx);
+ if (ret < 0) {
+ /*
+ * If we are unable to send the stream to the thread, there is
+ * a big problem so just stop everything.
+ */
+ goto error;
+ }
+
+ /* Remove node from the channel stream list. */
+ cds_list_del(&stream->send_node);
+ }
+
+error:
+ return ret;
+}
+
+/*
+ * Write metadata to the given channel using ustctl to convert the string to
+ * the ringbuffer.
+ *
+ * Return 0 on success else a negative value.
+ */
+static int push_metadata(struct lttng_consumer_channel *metadata,
+ const char *metadata_str, uint64_t target_offset, uint64_t len)
+{
+ int ret;
+
+ assert(metadata);
+ assert(metadata_str);
+
+ DBG("UST consumer writing metadata to channel %s", metadata->name);
+
+ assert(target_offset == metadata->contig_metadata_written);
+ ret = ustctl_write_metadata_to_channel(metadata->uchan, metadata_str, len);
+ if (ret < 0) {
+ ERR("ustctl write metadata fail with ret %d, len %ld", ret, len);
+ goto error;
+ }
+ metadata->contig_metadata_written += len;
+
+ ustctl_flush_buffer(metadata->metadata_stream->ustream, 1);
+
+error:
+ return ret;
+}
+
+/*
+ * Close metadata stream wakeup_fd using the given key to retrieve the channel.
+ *
+ * Return 0 on success else an LTTng error code.
+ */
+static int close_metadata(uint64_t chan_key)
+{
+ int ret;
+ struct lttng_consumer_channel *channel;
+
+ DBG("UST consumer close metadata key %lu", chan_key);
+
+ channel = consumer_find_channel(chan_key);
+ if (!channel) {
+ ERR("UST consumer close metadata %lu not found", chan_key);
+ ret = LTTNG_ERR_UST_CHAN_NOT_FOUND;
+ goto error;
+ }
+
+ ret = ustctl_stream_close_wakeup_fd(channel->metadata_stream->ustream);
+ if (ret < 0) {
+ ERR("UST consumer unable to close fd of metadata (ret: %d)", ret);
+ ret = LTTCOMM_CONSUMERD_ERROR_METADATA;
+ goto error;
+ }
+
+error:
+ return ret;
+}
+
+/*
+ * RCU read side lock MUST be acquired before calling this function.
+ *
+ * Return 0 on success else an LTTng error code.
+ */
+static int setup_metadata(struct lttng_consumer_local_data *ctx, uint64_t key)
+{
+ int ret;
+ struct lttng_consumer_channel *metadata;
+
+ DBG("UST consumer setup metadata key %lu", key);
+
+ metadata = consumer_find_channel(key);
+ if (!metadata) {
+ ERR("UST consumer push metadata %" PRIu64 " not found", key);
+ ret = LTTNG_ERR_UST_CHAN_NOT_FOUND;
+ goto error;
+ }
+
+ /*
+ * Send metadata stream to relayd if one available. Availability is
+ * known if the stream is still in the list of the channel.
+ */
+ if (cds_list_empty(&metadata->streams.head)) {
+ ERR("Metadata channel key %" PRIu64 ", no stream available.", key);
+ ret = LTTCOMM_CONSUMERD_ERROR_METADATA;
+ goto error;
+ }
+
+ /* Send metadata stream to relayd if needed. */
+ ret = send_stream_to_relayd(metadata->metadata_stream);
+ if (ret < 0) {
+ ret = LTTCOMM_CONSUMERD_ERROR_METADATA;
+ goto error;
+ }
+
+ ret = send_streams_to_thread(metadata, ctx);
+ if (ret < 0) {
+ /*
+ * If we are unable to send the stream to the thread, there is
+ * a big problem so just stop everything.
+ */
+ ret = LTTCOMM_CONSUMERD_FATAL;
+ goto error;
+ }
+ /* List MUST be empty after or else it could be reused. */
+ assert(cds_list_empty(&metadata->streams.head));
+
+ ret = 0;
+
+error:
+ return ret;
+}
+