x

src/core/BUILD

@@ -8274,6 +8274,7 @@ grpc_cc_library(
         "match_promise",
         "mpsc",
         "pipe",
+        "inter_activity_pipe",
         "//:promise",
     ],
 )

src/core/ext/transport/chaotic_good/frame_transport.h

@@ -66,12 +66,13 @@ class IncomingFrame {
 
 class FrameTransport : public RefCounted<FrameTransport> {
  public:
+  using ReadFramePipe = InterActivityPipe<IncomingFrame, 8>;
+
   // Spawn a read loop onto party - read frames from the wire, push them onto
   // frames.
   // TODO(ctiller): can likely use a buffered intra-party SpscSender here once
   // we write one.
-  virtual void StartReading(Party* party,
-                            InterActivityPipe<IncomingFrame>::Sender frames,
+  virtual void StartReading(Party* party, ReadFramePipe::Sender frames,
                             absl::AnyInvocable<void(absl::Status)> on_done) = 0;
   // Spawn a write loop onto party - write frames from frames to the wire.
   virtual void StartWriting(Party* party, MpscReceiver<Frame> frames,

src/core/ext/transport/chaotic_good/tcp_frame_transport.cc

@@ -227,7 +227,7 @@ auto TcpFrameTransport::ReadFrameBytes() {
 }
 
 void TcpFrameTransport::StartReading(
-    Party* party, MpscSender<IncomingFrame> frames,
+    Party* party, ReadFramePipe::Sender frames,
     absl::AnyInvocable<void(absl::Status)> on_done) {
   Crash("Not implemented");
 }

src/core/ext/transport/chaotic_good/tcp_frame_transport.h

@@ -71,8 +71,7 @@ class TcpFrameTransport final : public FrameTransport {
       std::shared_ptr<grpc_event_engine::experimental::EventEngine>
           event_engine);
 
-  void StartReading(Party* party,
-                    InterActivityPipe<IncomingFrame>::Sender frames,
+  void StartReading(Party* party, ReadFramePipe::Sender frames,
                     absl::AnyInvocable<void(absl::Status)> on_done) override;
   void StartWriting(Party* party, MpscReceiver<Frame> frames,
                     absl::AnyInvocable<void(absl::Status)> on_done) override;

test/core/transport/chaotic_good/BUILD

@@ -83,6 +83,7 @@ grpc_cc_test(
 
 grpc_cc_library(
     name = "mock_frame_transport",
+    testonly=1,
     srcs = [
         "mock_frame_transport.cc",
     ],
@@ -98,6 +99,7 @@ grpc_cc_library(
         "//src/core:inter_activity_latch",
         "//src/core:loop",
         "//src/core:try_seq",
+        "//test/core/promise:poll_matcher",
     ],
 )
 

test/core/transport/chaotic_good/mock_frame_transport.cc

@@ -18,6 +18,8 @@
 #include "src/core/lib/promise/loop.h"
 #include "src/core/lib/promise/race.h"
 #include "src/core/lib/promise/try_seq.h"
+#include "gmock/gmock.h"
+#include "test/core/promise/poll_matcher.h"
 
 namespace grpc_core {
 namespace chaotic_good {
@@ -37,7 +39,7 @@ MockFrameTransport::~MockFrameTransport() {
 }
 
 void MockFrameTransport::StartReading(
-    Party*, InterActivityPipe<IncomingFrame>::Sender frames,
+    Party*, ReadFramePipe::Sender frames,
     absl::AnyInvocable<void(absl::Status)> on_done) {
   reader_ = std::move(frames);
   on_read_done_ = std::move(on_done);
@@ -79,7 +81,7 @@ void MockFrameTransport::Read(Frame frame) {
   LOG(INFO) << "Read " << frame_interface.ToString();
   auto header = frame_interface.MakeHeader();
   frame_interface.SerializePayload(buffer);
-  reader_.UnbufferedImmediateSend(IncomingFrame(header, std::move(buffer)));
+  EXPECT_THAT(reader_.Push(IncomingFrame(header, std::move(buffer)))(), IsReady());
 }
 
 }  // namespace testing

test/core/transport/chaotic_good/mock_frame_transport.h

@@ -30,8 +30,7 @@ class MockFrameTransport final : public FrameTransport {
  public:
   ~MockFrameTransport() override;
 
-  void StartReading(Party* party,
-                    InterActivityPipe<IncomingFrame>::Sender frames,
+  void StartReading(Party* party, ReadFramePipe::Sender frames,
                     absl::AnyInvocable<void(absl::Status)> on_done) final;
   void StartWriting(Party* party, MpscReceiver<Frame> frames,
                     absl::AnyInvocable<void(absl::Status)> on_done) final;
@@ -52,7 +51,7 @@ class MockFrameTransport final : public FrameTransport {
   std::queue<ExpectedWrite> expected_writes_;
   std::shared_ptr<InterActivityLatch<absl::Status>> end_writes_ =
       std::make_shared<InterActivityLatch<absl::Status>>();
-  MpscSender<IncomingFrame> reader_;
+ReadFramePipe::Sender reader_;
   absl::AnyInvocable<void(absl::Status)> on_read_done_;
 };