Weave port [5/n]: Introduce weave message write request

PiperOrigin-RevId: 530915704

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",

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",
+    ],
+)

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 <algorithm>
+#include <string>
+
+#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<Packet> 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

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 <string>
+
+#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<Packet> 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<absl::Status> 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<absl::Status> write_request_status_;
+};
+
+}  // namespace weave
+}  // namespace nearby
+
+#endif  // THIRD_PARTY_NEARBY_INTERNAL_WEAVE_MESSAGE_WRITE_REQUEST_H_

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 <string>
+#include <type_traits>
+
+#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<MessageWriteRequest>::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<absl::Status> 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

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);