userspace socket: allow providing an existing PassthroughState when creating IoHandle pair

source/extensions/io_socket/user_space/io_handle.h

@@ -32,6 +32,7 @@ class PassthroughState {
 };
 
 using PassthroughStateSharedPtr = std::shared_ptr<PassthroughState>;
+using PassthroughStatePtr = std::unique_ptr<PassthroughState>;
 
 /**
  * The interface for the peer as a writer and supplied read status query.

source/extensions/io_socket/user_space/io_handle_impl.cc

@@ -392,6 +392,41 @@ void PassthroughStateImpl::mergeInto(envoy::config::core::v3::Metadata& metadata
   filter_state_objects_.clear();
   state_ = State::Done;
 }
+
+std::pair<IoHandleImplPtr, IoHandleImplPtr>
+IoHandleFactory::createIoHandlePair(PassthroughStatePtr state) {
+  PassthroughStateSharedPtr shared_state;
+  if (state != nullptr) {
+    shared_state = std::move(state);
+  } else {
+    shared_state = std::make_shared<PassthroughStateImpl>();
+  }
+  auto p = std::pair<IoHandleImplPtr, IoHandleImplPtr>{new IoHandleImpl(shared_state),
+                                                       new IoHandleImpl(shared_state)};
+  p.first->setPeerHandle(p.second.get());
+  p.second->setPeerHandle(p.first.get());
+  return p;
+}
+
+std::pair<IoHandleImplPtr, IoHandleImplPtr>
+IoHandleFactory::createBufferLimitedIoHandlePair(uint32_t buffer_size, PassthroughStatePtr state) {
+  PassthroughStateSharedPtr shared_state;
+  if (state != nullptr) {
+    shared_state = std::move(state);
+  } else {
+    shared_state = std::make_shared<PassthroughStateImpl>();
+  }
+  auto p = std::pair<IoHandleImplPtr, IoHandleImplPtr>{new IoHandleImpl(shared_state),
+                                                       new IoHandleImpl(shared_state)};
+  // This buffer watermark setting emulates the OS socket buffer parameter
+  // `/proc/sys/net/ipv4/tcp_{r,w}mem`.
+  p.first->setWatermarks(buffer_size);
+  p.second->setWatermarks(buffer_size);
+  p.first->setPeerHandle(p.second.get());
+  p.second->setPeerHandle(p.first.get());
+  return p;
+}
+
 } // namespace UserSpace
 } // namespace IoSocket
 } // namespace Extensions

source/extensions/io_socket/user_space/io_handle_impl.h

@@ -193,7 +193,7 @@ class PassthroughStateImpl : public PassthroughState, public Logger::Loggable<Lo
   void mergeInto(envoy::config::core::v3::Metadata& metadata,
                  StreamInfo::FilterState& filter_state) override;
 
-private:
+protected:
   enum class State { Created, Initialized, Done };
   State state_{State::Created};
   std::unique_ptr<envoy::config::core::v3::Metadata> metadata_;
@@ -203,27 +203,22 @@ class PassthroughStateImpl : public PassthroughState, public Logger::Loggable<Lo
 using IoHandleImplPtr = std::unique_ptr<IoHandleImpl>;
 class IoHandleFactory {
 public:
-  static std::pair<IoHandleImplPtr, IoHandleImplPtr> createIoHandlePair() {
-    auto state = std::make_shared<PassthroughStateImpl>();
-    auto p = std::pair<IoHandleImplPtr, IoHandleImplPtr>{new IoHandleImpl(state),
-                                                         new IoHandleImpl(state)};
-    p.first->setPeerHandle(p.second.get());
-    p.second->setPeerHandle(p.first.get());
-    return p;
-  }
+  /**
+   * @return a pair of connected IoHandleImpl instances.
+   * @param state optional existing value to use as the shared PassthroughState. If omitted, a
+   * newly constructed PassthroughStateImpl will be used.
+   */
   static std::pair<IoHandleImplPtr, IoHandleImplPtr>
-  createBufferLimitedIoHandlePair(uint32_t buffer_size) {
-    auto state = std::make_shared<PassthroughStateImpl>();
-    auto p = std::pair<IoHandleImplPtr, IoHandleImplPtr>{new IoHandleImpl(state),
-                                                         new IoHandleImpl(state)};
-    // This buffer watermark setting emulates the OS socket buffer parameter
-    // `/proc/sys/net/ipv4/tcp_{r,w}mem`.
-    p.first->setWatermarks(buffer_size);
-    p.second->setWatermarks(buffer_size);
-    p.first->setPeerHandle(p.second.get());
-    p.second->setPeerHandle(p.first.get());
-    return p;
-  }
+  createIoHandlePair(PassthroughStatePtr state = nullptr);
+
+  /**
+   * @return a pair of connected IoHandleImpl instances with pre-configured watermarks.
+   * @param buffer_size buffer watermark size in bytes
+   * @param state optional existing value to use as the shared PassthroughState. If omitted, a
+   * newly constructed PassthroughStateImpl will be used.
+   */
+  static std::pair<IoHandleImplPtr, IoHandleImplPtr>
+  createBufferLimitedIoHandlePair(uint32_t buffer_size, PassthroughStatePtr state = nullptr);
 };
 } // namespace UserSpace
 } // namespace IoSocket

test/extensions/io_socket/user_space/io_handle_impl_test.cc

@@ -1293,6 +1293,28 @@ TEST_F(IoHandleImplNotImplementedTest, ErrorOnGetOption) {
 TEST_F(IoHandleImplNotImplementedTest, ErrorOnIoctl) {
   EXPECT_THAT(io_handle_->ioctl(0, nullptr, 0, nullptr, 0, nullptr), IsNotSupportedResult());
 }
+
+class TestPassthroughState : public PassthroughStateImpl {};
+
+TEST(IoHandleFactoryTest, UseExistingPassthroughState) {
+  {
+    auto [io_handle, io_handle_peer] =
+        IoHandleFactory::createIoHandlePair(std::make_unique<TestPassthroughState>());
+    EXPECT_NE(std::dynamic_pointer_cast<TestPassthroughState>(io_handle->passthroughState()),
+              nullptr);
+    EXPECT_NE(std::dynamic_pointer_cast<TestPassthroughState>(io_handle_peer->passthroughState()),
+              nullptr);
+  }
+  {
+    auto [io_handle, io_handle_peer] = IoHandleFactory::createBufferLimitedIoHandlePair(
+        1024, std::make_unique<TestPassthroughState>());
+    EXPECT_NE(std::dynamic_pointer_cast<TestPassthroughState>(io_handle->passthroughState()),
+              nullptr);
+    EXPECT_NE(std::dynamic_pointer_cast<TestPassthroughState>(io_handle_peer->passthroughState()),
+              nullptr);
+  }
+}
+
 } // namespace
 } // namespace UserSpace
 } // namespace IoSocket