Create an abstract QmuxTransport

Summary: See title

Reviewed By: kvtsoy

Differential Revision: D114155798

fbshipit-source-id: 2555fbfd0950803b6b18c34595e61494d74d482a

Author: Aman Sharma

proxygen/lib/transport/qmux/CMakeLists.txt

@@ -33,18 +33,36 @@ proxygen_add_library(proxygen_transport_qmux_qmux_codec
     Folly::folly_io_iobuf
 )
 
+proxygen_add_library(proxygen_transport_qmux_qmux_transport
+  SRCS
+    QmuxTransport.cpp
+  EXPORTED_DEPS
+    Folly::folly_coro_task
+    Folly::folly_io_iobuf
+    Folly::folly_network_address
+)
+
+proxygen_add_library(proxygen_transport_qmux_folly_qmux_transport
+  SRCS
+    FollyQmuxTransport.cpp
+  DEPS
+    Folly::folly_io_coro_socket
+  EXPORTED_DEPS
+    proxygen_transport_qmux_qmux_transport
+)
+
 proxygen_add_library(proxygen_transport_qmux_qmux_session
   SRCS
     QmuxSession.cpp
   DEPS
     proxygen_transport_qmux_qmux_codec
-    Folly::folly_io_coro_socket
     Folly::folly_logging_logging
   EXPORTED_DEPS
     proxygen_http_coro_util_CoroWtSession
     proxygen_http_webtransport_wt_stream_manager
     proxygen_http_webtransport_wt_util
     proxygen_transport_qmux_qmux_framer
+    proxygen_transport_qmux_qmux_transport
     Folly::folly_io_async_async_base
 )
 
@@ -55,14 +73,14 @@ proxygen_add_library(proxygen_transport_qmux_qmux_connector
     mvfst::mvfst_folly_utils
     Folly::folly_coro_timeout
     Folly::folly_futures_core
-    Folly::folly_io_coro_socket
     Folly::folly_io_iobuf
     Folly::folly_logging_logging
     fmt::fmt
   EXPORTED_DEPS
     proxygen_http_webtransport_wt_stream_manager
     proxygen_transport_qmux_qmux_framer
     proxygen_transport_qmux_qmux_session
+    proxygen_transport_qmux_qmux_transport
     Folly::folly_coro_task
 )
 

proxygen/lib/transport/qmux/FollyQmuxTransport.cpp

@@ -0,0 +1,52 @@
+/*
+ * Copyright (c) Meta Platforms, Inc. and affiliates.
+ * All rights reserved.
+ *
+ * This source code is licensed under the BSD-style license found in the
+ * LICENSE file in the root directory of this source tree.
+ */
+
+#include <proxygen/lib/transport/qmux/FollyQmuxTransport.h>
+
+#include <folly/io/coro/Transport.h>
+
+namespace proxygen::qmux {
+
+FollyQmuxTransport::FollyQmuxTransport(
+    std::unique_ptr<folly::coro::TransportIf> transport)
+    : transport_(std::move(transport)) {
+}
+
+FollyQmuxTransport::~FollyQmuxTransport() = default;
+
+folly::coro::Task<size_t> FollyQmuxTransport::read(
+    folly::IOBufQueue& readBuf,
+    size_t minReadSize,
+    size_t newAllocationSize,
+    std::chrono::milliseconds timeout) {
+  return transport_->read(readBuf, minReadSize, newAllocationSize, timeout);
+}
+
+folly::coro::Task<folly::Unit> FollyQmuxTransport::write(
+    folly::IOBufQueue& writeBuf, std::chrono::milliseconds timeout) {
+  return transport_->write(writeBuf, timeout);
+}
+
+void FollyQmuxTransport::shutdownWrite() {
+  transport_->shutdownWrite();
+}
+
+folly::SocketAddress FollyQmuxTransport::getLocalAddress() const noexcept {
+  return transport_->getLocalAddress();
+}
+
+folly::SocketAddress FollyQmuxTransport::getPeerAddress() const noexcept {
+  return transport_->getPeerAddress();
+}
+
+folly::AsyncTransport* FollyQmuxTransport::getUnderlyingTransport()
+    const noexcept {
+  return transport_->getTransport();
+}
+
+} // namespace proxygen::qmux

proxygen/lib/transport/qmux/FollyQmuxTransport.h

@@ -0,0 +1,45 @@
+/*
+ * Copyright (c) Meta Platforms, Inc. and affiliates.
+ * All rights reserved.
+ *
+ * This source code is licensed under the BSD-style license found in the
+ * LICENSE file in the root directory of this source tree.
+ */
+
+#pragma once
+
+#include <memory>
+#include <proxygen/lib/transport/qmux/QmuxTransport.h>
+
+namespace folly::coro {
+class TransportIf;
+}
+
+namespace proxygen::qmux {
+
+class FollyQmuxTransport : public QmuxTransport {
+ public:
+  explicit FollyQmuxTransport(
+      std::unique_ptr<folly::coro::TransportIf> transport);
+  ~FollyQmuxTransport() override;
+
+  folly::coro::Task<size_t> read(folly::IOBufQueue& readBuf,
+                                 size_t minReadSize,
+                                 size_t newAllocationSize,
+                                 std::chrono::milliseconds timeout) override;
+
+  folly::coro::Task<folly::Unit> write(
+      folly::IOBufQueue& writeBuf, std::chrono::milliseconds timeout) override;
+
+  void shutdownWrite() override;
+
+  [[nodiscard]] folly::SocketAddress getLocalAddress() const noexcept override;
+  [[nodiscard]] folly::SocketAddress getPeerAddress() const noexcept override;
+  [[nodiscard]] folly::AsyncTransport* getUnderlyingTransport()
+      const noexcept override;
+
+ private:
+  std::unique_ptr<folly::coro::TransportIf> transport_;
+};
+
+} // namespace proxygen::qmux

proxygen/lib/transport/qmux/QmuxConnector.cpp

@@ -14,7 +14,6 @@
 #include <folly/futures/ThreadWheelTimekeeper.h>
 #include <folly/io/Cursor.h>
 #include <folly/io/IOBufQueue.h>
-#include <folly/io/coro/Transport.h>
 #include <folly/logging/xlog.h>
 #include <quic/folly_utils/Utils.h>
 #include <stdexcept>
@@ -146,7 +145,7 @@ PeelOutcome peelTransportParams(folly::IOBufQueue& buf) {
 }
 
 folly::coro::Task<QxTransportParams> readPeerTransportParams(
-    folly::coro::TransportIf& transport,
+    QmuxTransport& transport,
     folly::IOBufQueue& ingressBuf,
     std::chrono::milliseconds timeout) {
   while (true) {
@@ -198,7 +197,7 @@ folly::coro::Task<QmuxSession::Ptr> QmuxConnector::connect(
     folly::EventBase* evb,
     WtDir dir,
     QxTransportParams selfParams,
-    std::unique_ptr<folly::coro::TransportIf> transport,
+    std::unique_ptr<QmuxTransport> transport,
     std::chrono::milliseconds timeout,
     QmuxSession::Config sessionConfig) {
   XLOG(DBG4) << "QmuxConnector::connect dir=" << static_cast<int>(dir)

proxygen/lib/transport/qmux/QmuxConnector.h

@@ -13,15 +13,12 @@
 #include <proxygen/lib/http/webtransport/WtStreamManager.h>
 #include <proxygen/lib/transport/qmux/QmuxFramer.h>
 #include <proxygen/lib/transport/qmux/QmuxSession.h>
+#include <proxygen/lib/transport/qmux/QmuxTransport.h>
 
 namespace folly {
 class EventBase;
 } // namespace folly
 
-namespace folly::coro {
-class TransportIf;
-} // namespace folly::coro
-
 namespace proxygen::qmux {
 
 class QmuxConnector {
@@ -30,7 +27,7 @@ class QmuxConnector {
       folly::EventBase* evb,
       WtDir dir,
       QxTransportParams selfParams,
-      std::unique_ptr<folly::coro::TransportIf> transport,
+      std::unique_ptr<QmuxTransport> transport,
       std::chrono::milliseconds timeout,
       QmuxSession::Config sessionConfig = {});
 };

proxygen/lib/transport/qmux/QmuxSession.cpp

@@ -8,7 +8,6 @@
 
 #include <proxygen/lib/transport/qmux/QmuxSession.h>
 
-#include <folly/io/coro/Transport.h>
 #include <folly/logging/xlog.h>
 #include <proxygen/lib/transport/qmux/QmuxCodec.h>
 
@@ -138,7 +137,7 @@ class QmuxCallback : public QmuxCodec::Callback {
 QmuxSession::QmuxSession(folly::EventBase* evb,
                          WtDir dir,
                          QxTransportParams selfParams,
-                         std::unique_ptr<folly::coro::TransportIf> transport,
+                         std::unique_ptr<QmuxTransport> transport,
                          WtStreamManager::WtConfig wtConfig,
                          uint64_t peerMaxRecordSize,
                          uint64_t effectiveMaxIdleTimeoutMs,
@@ -169,7 +168,7 @@ proxygen::detail::WtExpected<folly::Unit>::Type QmuxSession::closeSession(
 }
 
 folly::AsyncTransport* QmuxSession::getUnderlyingTransport() const noexcept {
-  return transport_ ? transport_->getTransport() : nullptr;
+  return transport_ ? transport_->getUnderlyingTransport() : nullptr;
 }
 
 proxygen::detail::WtExpected<folly::Unit>::Type QmuxSession::sendDatagram(

proxygen/lib/transport/qmux/QmuxSession.h

@@ -14,15 +14,12 @@
 #include <proxygen/lib/http/webtransport/WtStreamManager.h>
 #include <proxygen/lib/http/webtransport/WtUtils.h>
 #include <proxygen/lib/transport/qmux/QmuxFramer.h>
+#include <proxygen/lib/transport/qmux/QmuxTransport.h>
 
 namespace folly {
 class AsyncTransport;
 }
 
-namespace folly::coro {
-class TransportIf;
-}
-
 namespace proxygen::qmux {
 
 using WtStreamManager = proxygen::detail::WtStreamManager;
@@ -43,7 +40,7 @@ class QmuxSession
   QmuxSession(folly::EventBase* evb,
               WtDir dir,
               QxTransportParams selfParams,
-              std::unique_ptr<folly::coro::TransportIf> transport,
+              std::unique_ptr<QmuxTransport> transport,
               WtStreamManager::WtConfig wtConfig,
               uint64_t peerMaxRecordSize,
               uint64_t effectiveMaxIdleTimeoutMs,
@@ -110,7 +107,7 @@ class QmuxSession
   folly::SocketAddress localAddr_;
   folly::SocketAddress peerAddr_;
   folly::CancellationSource cs_;
-  std::unique_ptr<folly::coro::TransportIf> transport_;
+  std::unique_ptr<QmuxTransport> transport_;
   QxTransportParams selfParams_;
   uint64_t effectiveMaxIdleTimeoutMs_;
   std::unique_ptr<folly::IOBuf> initialIngress_;

proxygen/lib/transport/qmux/QmuxTransport.cpp

@@ -0,0 +1,15 @@
+/*
+ * Copyright (c) Meta Platforms, Inc. and affiliates.
+ * All rights reserved.
+ *
+ * This source code is licensed under the BSD-style license found in the
+ * LICENSE file in the root directory of this source tree.
+ */
+
+#include <proxygen/lib/transport/qmux/QmuxTransport.h>
+
+namespace proxygen::qmux {
+
+QmuxTransport::~QmuxTransport() = default;
+
+} // namespace proxygen::qmux

proxygen/lib/transport/qmux/QmuxTransport.h

@@ -0,0 +1,43 @@
+/*
+ * Copyright (c) Meta Platforms, Inc. and affiliates.
+ * All rights reserved.
+ *
+ * This source code is licensed under the BSD-style license found in the
+ * LICENSE file in the root directory of this source tree.
+ */
+
+#pragma once
+
+#include <folly/SocketAddress.h>
+#include <folly/coro/Task.h>
+#include <folly/io/IOBufQueue.h>
+
+namespace folly {
+class AsyncTransport;
+}
+
+namespace proxygen::qmux {
+
+class QmuxTransport {
+ public:
+  virtual ~QmuxTransport();
+
+  virtual folly::coro::Task<size_t> read(folly::IOBufQueue& readBuf,
+                                         size_t minReadSize,
+                                         size_t newAllocationSize,
+                                         std::chrono::milliseconds timeout) = 0;
+
+  virtual folly::coro::Task<folly::Unit> write(
+      folly::IOBufQueue& writeBuf, std::chrono::milliseconds timeout) = 0;
+
+  virtual void shutdownWrite() = 0;
+
+  [[nodiscard]] virtual folly::SocketAddress getLocalAddress()
+      const noexcept = 0;
+  [[nodiscard]] virtual folly::SocketAddress getPeerAddress()
+      const noexcept = 0;
+  [[nodiscard]] virtual folly::AsyncTransport* getUnderlyingTransport()
+      const noexcept = 0;
+};
+
+} // namespace proxygen::qmux

proxygen/lib/transport/qmux/test/CMakeLists.txt

@@ -40,6 +40,7 @@ proxygen_add_test(TARGET QmuxSessionTest
   DEPENDS
     proxygen_transport_qmux_qmux_session
     proxygen_transport_qmux_qmux_codec
+    proxygen_transport_qmux_folly_qmux_transport
     proxygen_transport_qmux_qmux_framer
     proxygen_http_codec_webtransport_webtransport_framer
     proxygen_http_webtransport_wt_stream_manager
@@ -53,6 +54,7 @@ proxygen_add_test(TARGET QmuxConnectorTest
     QmuxConnectorTest.cpp
   DEPENDS
     proxygen_transport_qmux_qmux_connector
+    proxygen_transport_qmux_folly_qmux_transport
     proxygen_transport_qmux_qmux_framer
     TestCoroTransport
     testmain