+/*
+ * Allocate and return a consumer relayd socket.
+ */
+struct consumer_relayd_sock_pair *consumer_allocate_relayd_sock_pair(
+ int net_seq_idx)
+{
+ struct consumer_relayd_sock_pair *obj = NULL;
+
+ /* Negative net sequence index is a failure */
+ if (net_seq_idx < 0) {
+ goto error;
+ }
+
+ obj = zmalloc(sizeof(struct consumer_relayd_sock_pair));
+ if (obj == NULL) {
+ PERROR("zmalloc relayd sock");
+ goto error;
+ }
+
+ obj->net_seq_idx = net_seq_idx;
+ obj->refcount = 0;
+ lttng_ht_node_init_ulong(&obj->node, obj->net_seq_idx);
+ pthread_mutex_init(&obj->ctrl_sock_mutex, NULL);
+
+error:
+ return obj;
+}
+
+/*
+ * Find a relayd socket pair in the global consumer data.
+ *
+ * Return the object if found else NULL.
+ */
+struct consumer_relayd_sock_pair *consumer_find_relayd(int key)
+{
+ struct lttng_ht_iter iter;
+ struct lttng_ht_node_ulong *node;
+ struct consumer_relayd_sock_pair *relayd = NULL;
+
+ /* Negative keys are lookup failures */
+ if (key < 0) {
+ goto error;
+ }
+
+ rcu_read_lock();
+
+ lttng_ht_lookup(consumer_data.relayd_ht, (void *)((unsigned long) key),
+ &iter);
+ node = lttng_ht_iter_get_node_ulong(&iter);
+ if (node != NULL) {
+ relayd = caa_container_of(node, struct consumer_relayd_sock_pair, node);
+ }
+
+ rcu_read_unlock();
+
+error:
+ return relayd;
+}
+
+/*
+ * Handle stream for relayd transmission if the stream applies for network
+ * streaming where the net sequence index is set.
+ *
+ * Return destination file descriptor or negative value on error.
+ */
+int consumer_handle_stream_before_relayd(struct lttng_consumer_stream *stream,
+ size_t data_size)
+{
+ int outfd = -1, ret;
+ struct consumer_relayd_sock_pair *relayd;
+ struct lttcomm_relayd_data_hdr data_hdr;
+
+ /* Safety net */
+ assert(stream);
+
+ /* Reset data header */
+ memset(&data_hdr, 0, sizeof(data_hdr));
+
+ /* Get relayd reference of the stream. */
+ relayd = consumer_find_relayd(stream->net_seq_idx);
+ if (relayd == NULL) {
+ /* Stream is either local or corrupted */
+ goto error;
+ }
+
+ DBG("Consumer found relayd socks with index %d", stream->net_seq_idx);
+ if (stream->metadata_flag) {
+ /* Caller MUST acquire the relayd control socket lock */
+ ret = relayd_send_metadata(&relayd->control_sock, data_size);
+ if (ret < 0) {
+ goto error;
+ }
+
+ /* Metadata are always sent on the control socket. */
+ outfd = relayd->control_sock.fd;
+ } else {
+ /* Set header with stream information */
+ data_hdr.stream_id = htobe64(stream->relayd_stream_id);
+ data_hdr.data_size = htobe32(data_size);
+ /* Other fields are zeroed previously */
+
+ ret = relayd_send_data_hdr(&relayd->data_sock, &data_hdr,
+ sizeof(data_hdr));
+ if (ret < 0) {
+ goto error;
+ }
+
+ /* Set to go on data socket */
+ outfd = relayd->data_sock.fd;
+ }
+
+error:
+ return outfd;
+}
+