diff --git a/src/c/core/session/session.c b/src/c/core/session/session.c index 45900b1a0..2de8d0608 100644 --- a/src/c/core/session/session.c +++ b/src/c/core/session/session.c @@ -629,7 +629,11 @@ void uxr_flash_output_streams( while (uxr_prepare_next_reliable_buffer_to_send(stream, &buffer, &length, &seq_num)) { uxr_stamp_session_header(&session->info, id.raw, seq_num, buffer); - send_message(session, buffer, length); + if (!send_message(session, buffer, length)) + { + uxr_cancel_reliable_buffer_to_send(stream, seq_num); + break; + } } UXR_UNLOCK_STREAM_ID(session, id); diff --git a/src/c/core/session/stream/output_reliable_stream.c b/src/c/core/session/stream/output_reliable_stream.c index 93ca6e25a..ce5402226 100644 --- a/src/c/core/session/stream/output_reliable_stream.c +++ b/src/c/core/session/stream/output_reliable_stream.c @@ -192,6 +192,19 @@ bool uxr_prepare_next_reliable_buffer_to_send( return data_to_send; } +void uxr_cancel_reliable_buffer_to_send( + uxrOutputReliableStream* stream, + uxrSeqNum seq_num) +{ + uxrSeqNum next_seq_num = uxr_seq_num_add(seq_num, 1); + stream->last_sent = uxr_seq_num_sub(seq_num, 1); + if (stream->last_written == next_seq_num && + stream->offset == uxr_get_reliable_buffer_size(&stream->base, next_seq_num)) + { + stream->last_written = seq_num; + } +} + bool uxr_update_output_stream_heartbeat_timestamp( uxrOutputReliableStream* stream, int64_t current_timestamp) diff --git a/src/c/core/session/stream/output_reliable_stream_internal.h b/src/c/core/session/stream/output_reliable_stream_internal.h index 744b02f3d..7ba6ffa09 100644 --- a/src/c/core/session/stream/output_reliable_stream_internal.h +++ b/src/c/core/session/stream/output_reliable_stream_internal.h @@ -45,6 +45,9 @@ bool uxr_prepare_next_reliable_buffer_to_send( uint8_t** buffer, size_t* length, uxrSeqNum* seq_num); +void uxr_cancel_reliable_buffer_to_send( + uxrOutputReliableStream* stream, + uxrSeqNum seq_num); bool uxr_update_output_stream_heartbeat_timestamp( uxrOutputReliableStream* stream, diff --git a/test/unitary/session/Session.cpp b/test/unitary/session/Session.cpp index 9756c5428..7a6973179 100644 --- a/test/unitary/session/Session.cpp +++ b/test/unitary/session/Session.cpp @@ -103,6 +103,7 @@ class SessionTest : public testing::Test uint8_t input_reliable_buffer[MTU * HISTORY]; static int listening_counter; + static int sending_counter; static bool send_msg( void* instance, @@ -124,6 +125,13 @@ class SessionTest : public testing::Test EXPECT_EQ(size_t(MTU), len); return false; } + else if (std::string("FlashReliableStreamSendError") == + ::testing::UnitTest::GetInstance()->current_test_info()->name()) + { + EXPECT_EQ(size_t(OFFSET + SUBHEADER_SIZE + 8), len); + SessionTest::sending_counter++; + return false; + } else if (std::string("SendHeartbeat") == ::testing::UnitTest::GetInstance()->current_test_info()->name()) { EXPECT_EQ(size_t(HEARTBEAT_MAX_MSG_SIZE - (MAX_HEADER_SIZE - OFFSET)), len); @@ -272,6 +280,7 @@ class SessionTest : public testing::Test SessionTest* SessionTest::current = nullptr; int SessionTest::listening_counter; +int SessionTest::sending_counter; TEST_F(SessionTest, SetStatusCallback) { @@ -355,6 +364,28 @@ TEST_F(SessionTest, FlashStreams) uxr_flash_output_streams(&session); } +TEST_F(SessionTest, FlashReliableStreamSendError) +{ + SessionTest::sending_counter = 0; + + ucdrBuffer ub; + uxrStreamId output_reliable = uxr_stream_id(0, UXR_RELIABLE_STREAM, UXR_OUTPUT_STREAM); + (void) uxr_prepare_stream_to_write_submessage(&session, output_reliable, 8, &ub, 1, 0); + + uxrOutputReliableStream* stream = &session.streams.output_reliable[0]; + uxr_flash_output_streams(&session); + + EXPECT_EQ(1, SessionTest::sending_counter); + EXPECT_EQ(SEQ_NUM_MAX, stream->last_sent); + EXPECT_EQ(0u, stream->last_written); + + uxr_flash_output_streams(&session); + + EXPECT_EQ(2, SessionTest::sending_counter); + EXPECT_EQ(SEQ_NUM_MAX, stream->last_sent); + EXPECT_EQ(0u, stream->last_written); +} + TEST_F(SessionTest, WaitSessionStatusBad) { // The OK version is already checked with the CreateOk and DeleteOk test versions