diff --git a/app/src/WebSocketClient.cpp b/app/src/WebSocketClient.cpp index 3a76d9e..ae3a5ae 100644 --- a/app/src/WebSocketClient.cpp +++ b/app/src/WebSocketClient.cpp @@ -330,7 +330,7 @@ bool WebSocketClient::send_frame(uint8_t opcode, const uint8_t* payload, size_t frame.push_back(payload[i] ^ mask_key[i % 4]); } - return tx_queue_.enqueue(std::move(frame), opennow::websocket::WriteQueue::Clock::now()); + return tx_queue_.enqueue(std::move(frame)); } void WebSocketClient::send_message(const std::string& msg) { diff --git a/app/src/websocket_write_queue.hpp b/app/src/websocket_write_queue.hpp index 0058005..6004211 100644 --- a/app/src/websocket_write_queue.hpp +++ b/app/src/websocket_write_queue.hpp @@ -5,6 +5,7 @@ #include #include #include +#include #include #include @@ -35,12 +36,12 @@ class WriteQueue bytes <= MaximumBytes - queued_bytes_; } - bool enqueue(std::vector&& bytes, Clock::time_point now) + bool enqueue(std::vector&& bytes) { if (!can_enqueue(bytes.size())) return false; queued_bytes_ += bytes.size(); - frames_.push_back({std::move(bytes), 0, now + SendTimeout}); + frames_.push_back({std::move(bytes), 0, std::nullopt}); return true; } @@ -55,7 +56,9 @@ class WriteQueue { for (size_t writes = 0; writes < MaximumWritesPerPoll && !frames_.empty(); ++writes) { Frame& frame = frames_.front(); - if (now() >= frame.deadline) + if (!frame.deadline) + frame.deadline = now() + SendTimeout; + if (now() >= *frame.deadline) return DrainResult::TimedOut; const size_t length = std::min(MaximumWriteBytes, frame.bytes.size() - frame.offset); const WriteResult result = writer(frame.bytes.data() + frame.offset, length); @@ -79,7 +82,7 @@ class WriteQueue { std::vector bytes; size_t offset; - Clock::time_point deadline; + std::optional deadline; }; std::deque frames_; diff --git a/tests/websocket_write_queue_test.cpp b/tests/websocket_write_queue_test.cpp index 223dfef..5544058 100644 --- a/tests/websocket_write_queue_test.cpp +++ b/tests/websocket_write_queue_test.cpp @@ -44,8 +44,8 @@ int main() { Queue queue; FakeWriter writer; - assert(queue.enqueue({1, 2, 3, 4}, time)); - assert(queue.enqueue({5, 6}, time)); + assert(queue.enqueue({1, 2, 3, 4})); + assert(queue.enqueue({5, 6})); writer.results = {{Status::Progress, 2}, {Status::Again, 0}}; assert(queue.drain(writer, now) == Result::Pending); assert(writer.calls == 2); @@ -53,7 +53,7 @@ int main() writer.results = {{Status::Again, 0}}; assert(queue.drain(writer, now) == Result::Pending); assert(writer.calls == 3); - assert(queue.enqueue({7, 8}, time)); + assert(queue.enqueue({7, 8})); assert(queue.drain(writer, now) == Result::Complete); assert((writer.output == std::vector{1, 2, 3, 4, 5, 6, 7, 8})); } @@ -61,9 +61,9 @@ int main() { Queue queue; FakeWriter writer; - assert(queue.enqueue(std::vector(Queue::MaximumBytes, 9), time)); + assert(queue.enqueue(std::vector(Queue::MaximumBytes, 9))); assert(!queue.can_enqueue(1)); - assert(!queue.enqueue({1}, time)); + assert(!queue.enqueue({1})); assert(queue.drain(writer, now) == Result::Pending); assert(writer.calls == Queue::MaximumWritesPerPoll); assert(writer.output.size() == Queue::MaximumWritesPerPoll * Queue::MaximumWriteBytes); @@ -85,19 +85,19 @@ int main() Queue queue; FakeWriter writer; for (size_t i = 0; i < Queue::MaximumFrames; ++i) - assert(queue.enqueue({static_cast(i)}, time)); - assert(!queue.enqueue({65}, time)); + assert(queue.enqueue({static_cast(i)})); + assert(!queue.enqueue({65})); assert(queue.drain(writer, now) == Result::Pending); assert(writer.calls == Queue::MaximumWritesPerPoll); assert(queue.can_enqueue(1)); assert(!queue.can_enqueue(0)); - assert(!queue.enqueue({}, time)); + assert(!queue.enqueue({})); } { Queue queue; FakeWriter writer; - assert(queue.enqueue({1, 2, 3}, time)); + assert(queue.enqueue({1, 2, 3})); writer.results = {{Status::Progress, 1}, {Status::Again, 0}}; assert(queue.drain(writer, now) == Result::Pending); time += std::chrono::milliseconds(999); @@ -110,7 +110,7 @@ int main() assert((writer.output == std::vector{1, 2})); queue.reset(); writer = {}; - assert(queue.enqueue({4, 5}, time)); + assert(queue.enqueue({4, 5})); assert(queue.drain(writer, now) == Result::Complete); assert((writer.output == std::vector{4, 5})); } @@ -118,33 +118,37 @@ int main() { Queue queue; FakeWriter writer; - assert(queue.enqueue({1}, time)); - assert(queue.enqueue({2}, time)); - time += Queue::SendTimeout - std::chrono::milliseconds(1); + assert(queue.enqueue({1})); + assert(queue.enqueue({2})); + time += Queue::SendTimeout * 3; auto advancing_writer = [&](const uint8_t* data, size_t length) { const auto result = writer(data, length); - time += std::chrono::milliseconds(1); + time += Queue::SendTimeout - std::chrono::milliseconds(1); return result; }; - assert(queue.drain(advancing_writer, now) == Result::TimedOut); - assert(writer.calls == 1); - assert((writer.output == std::vector{1})); + assert(queue.drain(advancing_writer, now) == Result::Complete); + assert(writer.calls == 2); + assert((writer.output == std::vector{1, 2})); } { Queue queue; FakeWriter writer; - assert(queue.enqueue({1}, time)); + assert(queue.enqueue({1})); + assert(queue.enqueue({2})); + writer.results = {{Status::Again, 0}}; + assert(queue.drain(writer, now) == Result::Pending); time += Queue::SendTimeout; assert(queue.drain(writer, now) == Result::TimedOut); - assert(writer.calls == 0); + assert(writer.calls == 1); + assert(writer.output.empty()); } for (const auto status : {Status::Error, Status::Progress}) { Queue queue; FakeWriter writer; - assert(queue.enqueue({1, 2}, time)); - assert(queue.enqueue({3}, time)); + assert(queue.enqueue({1, 2})); + assert(queue.enqueue({3})); writer.results = {{status, 0}}; assert(queue.drain(writer, now) == (status == Status::Error ? Result::WriteError : Result::Stalled)); @@ -159,7 +163,7 @@ int main() { Queue queue; FakeWriter writer; - assert(queue.enqueue({1}, time)); + assert(queue.enqueue({1})); writer.results = {{Status::Progress, 2}}; assert(queue.drain(writer, now) == Result::WriteError); }