+
+/*
+ * Send a flush command to consumer using the given channel key.
+ *
+ * Return 0 on success else a negative value.
+ */
+int consumer_flush_channel(struct consumer_socket *socket, uint64_t key)
+{
+ int ret;
+ struct lttcomm_consumer_msg msg;
+
+ assert(socket);
+ assert(socket->fd >= 0);
+
+ DBG2("Consumer flush channel key %" PRIu64, key);
+
+ msg.cmd_type = LTTNG_CONSUMER_FLUSH_CHANNEL;
+ msg.u.flush_channel.key = key;
+
+ pthread_mutex_lock(socket->lock);
+ health_code_update();
+
+ ret = consumer_send_msg(socket, &msg);
+ if (ret < 0) {
+ goto end;
+ }
+
+end:
+ health_code_update();
+ pthread_mutex_unlock(socket->lock);
+ return ret;
+}
+
+/*
+ * Send a close metdata command to consumer using the given channel key.
+ *
+ * Return 0 on success else a negative value.
+ */
+int consumer_close_metadata(struct consumer_socket *socket,
+ uint64_t metadata_key)
+{
+ int ret;
+ struct lttcomm_consumer_msg msg;
+
+ assert(socket);
+ assert(socket->fd >= 0);
+
+ DBG2("Consumer close metadata channel key %" PRIu64, metadata_key);
+
+ msg.cmd_type = LTTNG_CONSUMER_CLOSE_METADATA;
+ msg.u.close_metadata.key = metadata_key;
+
+ pthread_mutex_lock(socket->lock);
+ health_code_update();
+
+ ret = consumer_send_msg(socket, &msg);
+ if (ret < 0) {
+ goto end;
+ }
+
+end:
+ health_code_update();
+ pthread_mutex_unlock(socket->lock);
+ return ret;
+}
+
+/*
+ * Send a setup metdata command to consumer using the given channel key.
+ *
+ * Return 0 on success else a negative value.
+ */
+int consumer_setup_metadata(struct consumer_socket *socket,
+ uint64_t metadata_key)
+{
+ int ret;
+ struct lttcomm_consumer_msg msg;
+
+ assert(socket);
+ assert(socket->fd >= 0);
+
+ DBG2("Consumer setup metadata channel key %" PRIu64, metadata_key);
+
+ msg.cmd_type = LTTNG_CONSUMER_SETUP_METADATA;
+ msg.u.setup_metadata.key = metadata_key;
+
+ pthread_mutex_lock(socket->lock);
+ health_code_update();
+
+ ret = consumer_send_msg(socket, &msg);
+ if (ret < 0) {
+ goto end;
+ }
+
+end:
+ health_code_update();
+ pthread_mutex_unlock(socket->lock);
+ return ret;
+}
+
+/*
+ * Send metadata string to consumer. Socket lock MUST be acquired.
+ *
+ * Return 0 on success else a negative value.
+ */
+int consumer_push_metadata(struct consumer_socket *socket,
+ uint64_t metadata_key, char *metadata_str, size_t len,
+ size_t target_offset)
+{
+ int ret;
+ struct lttcomm_consumer_msg msg;
+
+ assert(socket);
+ assert(socket->fd >= 0);
+
+ DBG2("Consumer push metadata to consumer socket %d", socket->fd);
+
+ msg.cmd_type = LTTNG_CONSUMER_PUSH_METADATA;
+ msg.u.push_metadata.key = metadata_key;
+ msg.u.push_metadata.target_offset = target_offset;
+ msg.u.push_metadata.len = len;
+
+ health_code_update();
+ ret = consumer_send_msg(socket, &msg);
+ if (ret < 0 || len == 0) {
+ goto end;
+ }
+
+ DBG3("Consumer pushing metadata on sock %d of len %zu", socket->fd, len);
+
+ ret = lttcomm_send_unix_sock(socket->fd, metadata_str, len);
+ if (ret < 0) {
+ goto end;
+ }
+
+ health_code_update();
+ ret = consumer_recv_status_reply(socket);
+ if (ret < 0) {
+ goto end;
+ }
+
+end:
+ health_code_update();
+ return ret;
+}