- if (!stream->tracefile_overwrite && new_id == 0) {
- stream->tracefile_overwrite = 1;
- }
- pthread_mutex_lock(&stream->viewer_stream_rotation_lock);
- if (stream->tracefile_overwrite) {
- stream->oldest_tracefile_id =
- (stream->oldest_tracefile_id + 1) %
- stream->tracefile_count;
+ stream->tracefile_size_current = 0;
+ conn->protocol.data.state.receive_payload.rotate_index = true;
+ }
+
+ ret = 0;
+end_stream_unlock:
+ pthread_mutex_unlock(&stream->lock);
+ stream_put(stream);
+end:
+ return status;
+}
+
+static enum relay_connection_status relay_process_data_receive_payload(
+ struct relay_connection *conn)
+{
+ int ret;
+ enum relay_connection_status status = RELAY_CONNECTION_STATUS_OK;
+ struct relay_stream *stream;
+ struct data_connection_state_receive_payload *state =
+ &conn->protocol.data.state.receive_payload;
+ const size_t chunk_size = RECV_DATA_BUFFER_SIZE;
+ char data_buffer[chunk_size];
+ bool partial_recv = false;
+ bool new_stream = false, close_requested = false;
+ uint64_t left_to_receive = state->left_to_receive;
+ struct relay_session *session;
+ int stream_fd = -1;
+
+ DBG3("Receiving data for stream id %" PRIu64 " seqnum %" PRIu64 ", %" PRIu64" bytes received, %" PRIu64 " bytes left to receive",
+ state->header.stream_id, state->header.net_seq_num,
+ state->received, left_to_receive);
+
+ stream = stream_get_by_id(state->header.stream_id);
+ if (!stream) {
+ /* Protocol error. */
+ ERR("relay_process_data_receive_payload: cannot find stream %" PRIu64,
+ state->header.stream_id);
+ status = RELAY_CONNECTION_STATUS_ERROR;
+ goto end;
+ }
+
+ pthread_mutex_lock(&stream->lock);
+ session = stream->trace->session;
+ if (!conn->session) {
+ ret = connection_set_session(conn, session);
+ if (ret) {
+ status = RELAY_CONNECTION_STATUS_ERROR;
+ goto end_stream_unlock;