From 69d2edaea95dbd7487c4f990776901bfccfad1b5 Mon Sep 17 00:00:00 2001 From: Anay Wadhera Date: Wed, 10 May 2023 08:31:40 -0700 Subject: [PATCH] Weave port [5/n]: Introduce weave message write request PiperOrigin-RevId: 530915704 --- Package.swift | 1 + internal/weave/BUILD | 18 ++++ internal/weave/message_write_request.cc | 60 +++++++++++ internal/weave/message_write_request.h | 77 ++++++++++++++ internal/weave/message_write_request_test.cc | 105 +++++++++++++++++++ internal/weave/packet.h | 2 +- 6 files changed, 262 insertions(+), 1 deletion(-) create mode 100644 internal/weave/message_write_request.cc create mode 100644 internal/weave/message_write_request.h create mode 100644 internal/weave/message_write_request_test.cc diff --git a/Package.swift b/Package.swift index a1980a1d..b690d341 100644 --- a/Package.swift +++ b/Package.swift @@ -574,6 +574,7 @@ let package = Package( "internal/test/fake_device_info_test.cc", "internal/test/fake_task_runner_test.cc", "internal/weave/control_packet_write_request_test.cc", + "internal/weave/message_write_request_test.cc", "internal/weave/packet_test.cc", "internal/weave/packet_sequence_number_generator_test.cc", "internal/weave/packetizer_test.cc", diff --git a/internal/weave/BUILD b/internal/weave/BUILD index d07faf54..7a431e4f 100644 --- a/internal/weave/BUILD +++ b/internal/weave/BUILD @@ -1,12 +1,14 @@ cc_library( name = "weave", srcs = [ + "message_write_request.cc", "packet.cc", "packet_sequence_number_generator.cc", "packetizer.cc", ], hdrs = [ "control_packet_write_request.h", + "message_write_request.h", "packet.h", "packet_sequence_number_generator.h", "packetizer.h", @@ -77,3 +79,19 @@ cc_test( "@com_google_googletest//:gtest_main", ], ) + +cc_test( + name = "message_write_request_test", + srcs = [ + "message_write_request_test.cc", + ], + deps = [ + ":weave", + "//internal/platform:types", + "//internal/platform/implementation/g3", # build_cleaner: keep + "@com_github_protobuf_matchers//protobuf-matchers", + "@com_google_absl//absl/status", + "@com_google_absl//absl/strings", + "@com_google_googletest//:gtest_main", + ], +) diff --git a/internal/weave/message_write_request.cc b/internal/weave/message_write_request.cc new file mode 100644 index 00000000..648c975a --- /dev/null +++ b/internal/weave/message_write_request.cc @@ -0,0 +1,60 @@ +// Copyright 2023 Google LLC +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// https://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +#include "internal/weave/message_write_request.h" + +#include +#include + +#include "absl/status/status.h" + +namespace nearby { +namespace weave { + +MessageWriteRequest::MessageWriteRequest(absl::string_view message) { + message_ = std::string(message); + position_ = 0; +} + +bool MessageWriteRequest::IsStarted() const { return position_ != 0; } + +bool MessageWriteRequest::IsFinished() const { + return position_ >= message_.size(); +} + +// max_packet_size is determined and checked at the socket level, so we only +// need to make sure we're getting a valid packet size > 0. We will take +// max_packet_size - 1 (due to the header) bytes from the message if available, +// otherwise taking the remaining bytes in the message and completing the +// request. +absl::StatusOr MessageWriteRequest::NextPacket(int max_packet_size) { + if (IsFinished()) { + return absl::OutOfRangeError( + "Message is finished sending, no more packets."); + } + if (max_packet_size <= 0) { + return absl::InvalidArgumentError( + "max_packet_size must be greater than 0."); + } + bool is_first = !IsStarted(); + int next_packet_len = std::min(max_packet_size - Packet::kPacketHeaderLength, + (int)message_.size() - position_); + auto next_packet_bytes = message_.substr(position_, next_packet_len); + position_ += next_packet_len; + return Packet::CreateDataPacket(is_first, IsFinished(), + ByteArray(next_packet_bytes)); +} + +} // namespace weave +} // namespace nearby diff --git a/internal/weave/message_write_request.h b/internal/weave/message_write_request.h new file mode 100644 index 00000000..6a225936 --- /dev/null +++ b/internal/weave/message_write_request.h @@ -0,0 +1,77 @@ +// Copyright 2023 Google LLC +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// https://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +#ifndef THIRD_PARTY_NEARBY_INTERNAL_WEAVE_MESSAGE_WRITE_REQUEST_H_ +#define THIRD_PARTY_NEARBY_INTERNAL_WEAVE_MESSAGE_WRITE_REQUEST_H_ + +#include + +#include "internal/platform/future.h" +#include "internal/weave/packet.h" + +namespace nearby { +namespace weave { + +// Implementation of WriteRequest interface that exists to serialize messages +// into Weave packets and track the sending progress. +// +// This class is not thread-safe, calls to this class will be serialized at the +// socket level. +class MessageWriteRequest { + public: + explicit MessageWriteRequest(absl::string_view message); + MessageWriteRequest(MessageWriteRequest&& other) = default; + MessageWriteRequest& operator=(MessageWriteRequest&& other) = default; + + bool IsStarted() const; + bool IsFinished() const; + + absl::StatusOr NextPacket(int max_packet_size); + + // Gets the future result, the socket will set this result in the future. + // The socket only sets the result once the request has been completed. + nearby::Future GetWriteStatusFuture() { + return write_request_status_; + } + + // This function sets the nearby::Future instance from GetResultFuture() and + // is called when the caller manages to successfully write the packets in + // this WriteRequest (as indicated by IsFinished()), or when the caller learns + // that a packet could not be sent across the wire due to an error or + // disconnection. + void SetWriteStatus(absl::Status status) { + write_request_status_.Set(status); + } + + // Two message write requests are the same if they have the same message, + // even if they have different progress members due to referencing and + // other quirks. + bool operator==(const MessageWriteRequest& other) const { + return message_ == other.message_; + } + + bool operator!=(const MessageWriteRequest& other) const { + return !(*this == other); + } + + private: + std::string message_; + int position_; + nearby::Future write_request_status_; +}; + +} // namespace weave +} // namespace nearby + +#endif // THIRD_PARTY_NEARBY_INTERNAL_WEAVE_MESSAGE_WRITE_REQUEST_H_ diff --git a/internal/weave/message_write_request_test.cc b/internal/weave/message_write_request_test.cc new file mode 100644 index 00000000..fac47d0b --- /dev/null +++ b/internal/weave/message_write_request_test.cc @@ -0,0 +1,105 @@ +// Copyright 2023 Google LLC +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// https://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +#include "internal/weave/message_write_request.h" + +#include +#include + +#include "gmock/gmock.h" +#include "protobuf-matchers/protocol-buffer-matchers.h" +#include "gtest/gtest.h" +#include "absl/status/status.h" +#include "absl/strings/string_view.h" +#include "internal/platform/future.h" +#include "internal/weave/packet.h" + +namespace nearby { +namespace weave { +namespace { + +constexpr absl::string_view kShortMessage = "short"; +constexpr absl::string_view kLongMessage = "This is a long message."; +constexpr absl::string_view kLongFirstHalf = "This is a long"; +constexpr absl::string_view kLongSecondHalf = " message."; + +TEST(MessageWriteRequestTest, WriteRequestIsNotTriviallyConstructible) { + EXPECT_FALSE(std::is_trivially_constructible::value); +} + +TEST(MessageWriteRequestTest, ShortWriteRequestWorks) { + MessageWriteRequest request = MessageWriteRequest(kShortMessage); + EXPECT_FALSE(request.IsFinished()); + EXPECT_FALSE(request.IsStarted()); + Packet packet = request.NextPacket(15).value(); + EXPECT_TRUE(packet.IsDataPacket()); + EXPECT_TRUE(packet.IsFirstPacket()); + EXPECT_TRUE(packet.IsLastPacket()); + EXPECT_EQ(packet.GetPayload(), kShortMessage); + EXPECT_TRUE(request.IsFinished()); + EXPECT_TRUE(request.IsStarted()); +} + +TEST(MessageWriteRequestTest, LongWriteRequestWorks) { + MessageWriteRequest request = MessageWriteRequest(kLongMessage); + EXPECT_FALSE(request.IsFinished()); + EXPECT_FALSE(request.IsStarted()); + Packet packet = request.NextPacket(15).value(); + EXPECT_TRUE(packet.IsDataPacket()); + EXPECT_TRUE(packet.IsFirstPacket()); + EXPECT_FALSE(packet.IsLastPacket()); + EXPECT_EQ(packet.GetPayload(), kLongFirstHalf); + EXPECT_FALSE(request.IsFinished()); + EXPECT_TRUE(request.IsStarted()); + Packet nextPacket = request.NextPacket(15).value(); + EXPECT_TRUE(nextPacket.IsDataPacket()); + EXPECT_FALSE(nextPacket.IsFirstPacket()); + EXPECT_TRUE(nextPacket.IsLastPacket()); + EXPECT_EQ(nextPacket.GetPayload(), kLongSecondHalf); + EXPECT_TRUE(request.IsFinished()); +} + +TEST(MessageWriteRequestTest, TestResourceExhaustionOnceMessageSent) { + MessageWriteRequest request = MessageWriteRequest(kShortMessage); + EXPECT_FALSE(request.IsFinished()); + EXPECT_FALSE(request.IsStarted()); + Packet packet = request.NextPacket(15).value(); + EXPECT_TRUE(packet.IsDataPacket()); + EXPECT_TRUE(packet.IsFirstPacket()); + EXPECT_TRUE(packet.IsLastPacket()); + EXPECT_EQ(packet.GetPayload(), kShortMessage); + EXPECT_TRUE(request.IsFinished()); + EXPECT_TRUE(request.IsStarted()); + EXPECT_THAT(request.NextPacket(15), + testing::status::StatusIs(absl::StatusCode::kOutOfRange)); +} + +TEST(MessageWriteRequestTest, TestGetSetFuture) { + MessageWriteRequest request = MessageWriteRequest(kShortMessage); + nearby::Future result = request.GetWriteStatusFuture(); + request.SetWriteStatus(absl::InternalError("")); + EXPECT_THAT(result.Get().GetResult(), + testing::status::StatusIs(absl::StatusCode::kInternal)); +} + +TEST(MessageWriteRequestTest, TestInvalidPacketSize) { + MessageWriteRequest request = MessageWriteRequest(kShortMessage); + auto result = request.NextPacket(0); + EXPECT_THAT(result, + testing::status::StatusIs(absl::StatusCode::kInvalidArgument)); +} + +} // namespace +} // namespace weave +} // namespace nearby diff --git a/internal/weave/packet.h b/internal/weave/packet.h index b92d2e28..3597a1dd 100644 --- a/internal/weave/packet.h +++ b/internal/weave/packet.h @@ -82,6 +82,7 @@ packet size. class Packet { public: static constexpr int kMaxPacketCounter = 0b111; + static constexpr int kPacketHeaderLength = 1; enum class ControlPacketType { kControlConnectionRequest = 0, kControlConnectionConfirm = 1, @@ -134,7 +135,6 @@ class Packet { static constexpr char kFirstPacketBit = 0b00001000; static constexpr char kLastPacketBit = 0b00000100; static constexpr char kMaskType = 0b10000000; - static constexpr int kPacketHeaderLength = 1; static Packet CreateControlPacket(ControlPacketType command_number, int payload_size);