+ * Receive an index for a specific stream.
+ *
+ * Return 0 on success else a negative value.
+ */
+static int relay_recv_index(struct lttcomm_relayd_hdr *recv_hdr,
+ struct relay_connection *conn)
+{
+ int ret, send_ret;
+ struct relay_session *session = conn->session;
+ struct lttcomm_relayd_index index_info;
+ struct relay_index *index;
+ struct lttcomm_relayd_generic_reply reply;
+ struct relay_stream *stream;
+ uint64_t net_seq_num;
+
+ assert(conn);
+
+ DBG("Relay receiving index");
+
+ if (!session || conn->version_check_done == 0) {
+ ERR("Trying to close a stream before version check");
+ ret = -1;
+ goto end_no_session;
+ }
+
+ ret = conn->sock->ops->recvmsg(conn->sock, &index_info,
+ sizeof(index_info), 0);
+ if (ret < sizeof(index_info)) {
+ if (ret == 0) {
+ /* Orderly shutdown. Not necessary to print an error. */
+ DBG("Socket %d did an orderly shutdown", conn->sock->fd);
+ } else {
+ ERR("Relay didn't receive valid index struct size : %d", ret);
+ }
+ ret = -1;
+ goto end_no_session;
+ }
+
+ net_seq_num = be64toh(index_info.net_seq_num);
+
+ stream = stream_get_by_id(be64toh(index_info.relay_stream_id));
+ if (!stream) {
+ ERR("stream_get_by_id not found");
+ ret = -1;
+ goto end;
+ }
+ pthread_mutex_lock(&stream->lock);
+
+ /* Live beacon handling */
+ if (index_info.packet_size == 0) {
+ DBG("Received live beacon for stream %" PRIu64,
+ stream->stream_handle);
+
+ /*
+ * Only flag a stream inactive when it has already
+ * received data and no indexes are in flight.
+ */
+ if (stream->total_index_received > 0
+ && stream->indexes_in_flight == 0) {
+ stream->beacon_ts_end =
+ be64toh(index_info.timestamp_end);
+ }
+ ret = 0;
+ goto end_stream_put;
+ } else {
+ stream->beacon_ts_end = -1ULL;
+ }
+
+ if (stream->ctf_stream_id == -1ULL) {
+ stream->ctf_stream_id = be64toh(index_info.stream_id);
+ }
+ index = relay_index_get_by_id_or_create(stream, net_seq_num);
+ if (!index) {
+ ret = -1;
+ ERR("relay_index_get_by_id_or_create index NULL");
+ goto end_stream_put;
+ }
+ if (set_index_control_data(index, &index_info)) {
+ ERR("set_index_control_data error");
+ relay_index_put(index);
+ ret = -1;
+ goto end_stream_put;
+ }
+ ret = relay_index_try_flush(index);
+ if (ret == 0) {
+ stream->total_index_received++;
+ } else if (ret > 0) {
+ /* no flush. */
+ ret = 0;
+ } else {
+ ERR("relay_index_try_flush error %d", ret);
+ relay_index_put(index);
+ ret = -1;
+ }
+
+end_stream_put:
+ pthread_mutex_unlock(&stream->lock);
+ stream_put(stream);
+
+end:
+
+ memset(&reply, 0, sizeof(reply));
+ if (ret < 0) {
+ reply.ret_code = htobe32(LTTNG_ERR_UNK);
+ } else {
+ reply.ret_code = htobe32(LTTNG_OK);
+ }
+ send_ret = conn->sock->ops->sendmsg(conn->sock, &reply, sizeof(reply), 0);
+ if (send_ret < 0) {
+ ERR("Relay sending close index id reply");
+ ret = send_ret;
+ }
+
+end_no_session:
+ return ret;
+}
+
+/*
+ * Receive the streams_sent message.
+ *
+ * Return 0 on success else a negative value.
+ */
+static int relay_streams_sent(struct lttcomm_relayd_hdr *recv_hdr,
+ struct relay_connection *conn)
+{
+ int ret, send_ret;
+ struct lttcomm_relayd_generic_reply reply;
+
+ assert(conn);
+
+ DBG("Relay receiving streams_sent");
+
+ if (!conn->session || conn->version_check_done == 0) {
+ ERR("Trying to close a stream before version check");
+ ret = -1;
+ goto end_no_session;
+ }
+
+ /*
+ * Publish every pending stream in the connection recv list which are
+ * now ready to be used by the viewer.
+ */
+ publish_connection_local_streams(conn);
+
+ memset(&reply, 0, sizeof(reply));
+ reply.ret_code = htobe32(LTTNG_OK);
+ send_ret = conn->sock->ops->sendmsg(conn->sock, &reply, sizeof(reply), 0);
+ if (send_ret < 0) {
+ ERR("Relay sending sent_stream reply");
+ ret = send_ret;
+ } else {
+ /* Success. */
+ ret = 0;
+ }
+
+end_no_session:
+ return ret;
+}
+
+/*
+ * Process the commands received on the control socket