Full picture of in memory listener and connection

api/envoy/config/listener/v3/internal_listener.proto

@@ -0,0 +1,20 @@
+syntax = "proto3";
+
+package envoy.config.listener.v3;
+
+import "udpa/annotations/status.proto";
+import "udpa/annotations/versioning.proto";
+
+option java_package = "io.envoyproxy.envoy.config.listener.v3";
+option java_outer_classname = "InternalListenerProto";
+option java_multiple_files = true;
+option (udpa.annotations.file_status).package_version_status = ACTIVE;
+
+// [#not-implemented-hide:]
+// [#protodoc-title: internal listener]
+// Describes a type of internal listener which expects to serve the cluster in
+// the same envoy process.
+message InternalListener {
+  option (udpa.annotations.versioning).previous_message_type =
+      "envoy.config.listener.v2.InternalListener";
+}

api/envoy/config/listener/v3/listener.proto

@@ -8,6 +8,7 @@ import "envoy/config/core/v3/base.proto";
 import "envoy/config/core/v3/extension.proto";
 import "envoy/config/core/v3/socket_option.proto";
 import "envoy/config/listener/v3/api_listener.proto";
+import "envoy/config/listener/v3/internal_listener.proto";
 import "envoy/config/listener/v3/listener_components.proto";
 import "envoy/config/listener/v3/udp_listener_config.proto";
 
@@ -36,7 +37,7 @@ message ListenerCollection {
   repeated udpa.core.v1.CollectionEntry entries = 1;
 }
 
-// [#next-free-field: 25]
+// [#next-free-field: 26]
 message Listener {
   option (udpa.annotations.versioning).previous_message_type = "envoy.api.v2.Listener";
 
@@ -263,4 +264,10 @@ message Listener {
   // The maximum length a tcp listener's pending connections queue can grow to. If no value is
   // provided net.core.somaxconn will be used on Linux and 128 otherwise.
   google.protobuf.UInt32Value tcp_backlog_size = 24;
+
+  // Used to represent an internal listener, which accepts connection from the cluster in the same envoy process.
+  // When this field is set, the address field must be :ref:`envoy internal address
+  // <envoy_api_field_config.core.v3.EnvoyInternalAddress>`.
+  // [#not-implemented-hide:]
+  InternalListener internal_listener = 25;
 }

include/envoy/event/dispatcher.h

@@ -113,6 +113,16 @@ class Dispatcher {
                          Network::TransportSocketPtr&& transport_socket,
                          const Network::ConnectionSocket::OptionsSharedPtr& options) PURE;
 
+  /**
+   * Creates an client internal connection. Does NOT initiate the connection;
+   * the caller must then call connect() on the returned Network::ClientConnection.
+   * @param internal_address supplies the internal address to connect to.
+   * @param local_address supplies an address to bind to or nullptr if no bind is necessary.
+   * @return Network::ClientConnectionPtr a client connection that is owned by the caller.
+   */
+  virtual Network::ClientConnectionPtr
+  createInternalConnection(Network::Address::InstanceConstSharedPtr internal_address,
+                           Network::Address::InstanceConstSharedPtr local_address) PURE;
   /**
    * Creates an async DNS resolver. The resolver should only be used on the thread that runs this
    * dispatcher.

include/envoy/network/listener.h

@@ -146,6 +146,11 @@ class ListenerConfig {
    */
   virtual UdpPacketWriterFactoryOptRef udpPacketWriterFactory() PURE;
 
+  /**
+   * @return true if this listener is internal listener.
+   */
+  virtual bool isInternalListener() PURE;
+
   /**
    * @return the ``UdpListenerWorkerRouter`` for this listener. This will
    * be non-empty iff this is a UDP listener.
@@ -203,6 +208,18 @@ class TcpListenerCallbacks {
   virtual void onReject() PURE;
 };
 
+/**
+ * Callbacks invoked by a internal listener.
+ */
+class InternalListenerCallbacks {
+public:
+  virtual ~InternalListenerCallbacks() = default;
+
+  virtual void setupNewConnection(Network::ConnectionPtr server_conn,
+                                  Network::ConnectionSocketPtr socket) PURE;
+  virtual void onNewSocket(Network::ConnectionSocketPtr socket) PURE;
+};
+
 /**
  * Utility struct that encapsulates the information from a udp socket's recvmmsg call.
  */

source/common/event/BUILD

@@ -30,6 +30,7 @@ envoy_cc_library(
         "//source/common/common:assert_lib",
         "//source/common/common:thread_lib",
         "//source/common/filesystem:watcher_lib",
+        "//source/common/network:buffered_io_socket_handle_lib",
         "//source/common/network:connection_lib",
         "//source/common/network:dns_lib",
         "//source/common/network:listener_lib",

source/common/event/dispatcher_impl.cc

@@ -3,9 +3,12 @@
 #include <chrono>
 #include <cstdint>
 #include <functional>
+#include <memory>
 #include <string>
 #include <vector>
 
+#include "common/network/address_impl.h"
+#include "common/network/raw_buffer_socket.h"
 #include "envoy/api/api.h"
 #include "envoy/network/listen_socket.h"
 #include "envoy/network/listener.h"
@@ -18,6 +21,7 @@
 #include "common/event/signal_impl.h"
 #include "common/event/timer_impl.h"
 #include "common/filesystem/watcher_impl.h"
+#include "common/network/buffered_io_socket_handle_impl.h"
 #include "common/network/connection_impl.h"
 #include "common/network/dns_impl.h"
 #include "common/network/tcp_listener_impl.h"
@@ -117,6 +121,71 @@ DispatcherImpl::createClientConnection(Network::Address::InstanceConstSharedPtr
                                                          std::move(transport_socket), options);
 }
 
+namespace {
+Network::Address::InstanceConstSharedPtr
+nextClientAddress(const Network::Address::InstanceConstSharedPtr& server_address) {
+  uint64_t id = 0;
+  return std::make_shared<Network::Address::EnvoyInternalInstance>(absl::StrCat(server_address->asStringView(), "_", ++id));
+}
+} // namespace
+
+Network::ClientConnectionPtr
+DispatcherImpl::createInternalConnection(Network::Address::InstanceConstSharedPtr internal_address,
+                                         Network::Address::InstanceConstSharedPtr local_address) {
+  ASSERT(isThreadSafe());
+  if (internal_address == nullptr) {
+    return nullptr;
+  }
+  if (local_address == nullptr) {
+    local_address = nextClientAddress(internal_address);
+  }
+  // Find the internal listener callback. The listener will setup the server connection.
+  auto iter = internal_listeners_.find(internal_address->asString());
+  for (const auto& [name, _] : internal_listeners_) {
+    ENVOY_LOG_MISC(debug, "lambdai: p listener {}", name);
+  }
+  if (iter == internal_listeners_.end()) {
+    ENVOY_LOG_MISC(debug, "lambdai: no valid listener registered for envoy internal address {}",
+                   internal_address->asString());
+    return std::make_unique<Network::ClosingClientConnectionImpl>(*this, internal_address,
+                                                                  local_address);
+  }
+
+  auto client_io_handle_ = std::make_unique<Network::BufferedIoSocketHandleImpl>();
+  auto server_io_handle_ = std::make_unique<Network::BufferedIoSocketHandleImpl>();
+  client_io_handle_->setWritablePeer(server_io_handle_.get());
+  server_io_handle_->setWritablePeer(client_io_handle_.get());
+
+  Network::RawBufferSocketFactory client_transport_socket_factory;
+  // ConnectionSocket conn_socket
+  auto client_conn_socket = std::make_unique<Network::InternalConnectionSocketImpl>(
+      std::move(client_io_handle_), local_address, internal_address);
+  auto server_conn_socket = std::make_unique<Network::InternalConnectionSocketImpl>(
+      std::move(server_io_handle_), internal_address, local_address);
+  ENVOY_LOG_MISC(debug, "lambdai: internal address {}", internal_address->asString());
+  ENVOY_LOG_MISC(debug, "lambdai: client address {}", local_address->asString());
+
+  auto client_conn = std::make_unique<Network::ClientConnectionImpl>(
+      *this, internal_address, local_address,
+      client_transport_socket_factory.createTransportSocket(nullptr), nullptr,
+      std::move(client_conn_socket));
+
+  (iter->second)(internal_address, std::move(server_conn_socket));
+  return client_conn;
+}
+
+void DispatcherImpl::registerInternalListener(
+    const std::string& internal_listener_id,
+    DispatcherImpl::InternalConnectionCallback internal_conn_callback) {
+  if (internal_conn_callback == nullptr) {
+    ENVOY_LOG_MISC(debug, "lambdai: unregister pipe factory on address {}", internal_listener_id);
+    internal_listeners_.erase(internal_listener_id);
+  } else {
+    ENVOY_LOG_MISC(debug, "lambdai: register pipe factory on address {}", internal_listener_id);
+    internal_listeners_[internal_listener_id] = internal_conn_callback;
+  }
+}
+
 Network::DnsResolverSharedPtr DispatcherImpl::createDnsResolver(
     const std::vector<Network::Address::InstanceConstSharedPtr>& resolvers,
     const bool use_tcp_for_dns_lookups) {

source/common/event/dispatcher_impl.h

@@ -12,6 +12,7 @@
 #include "envoy/event/deferred_deletable.h"
 #include "envoy/event/dispatcher.h"
 #include "envoy/network/connection_handler.h"
+#include "envoy/network/listen_socket.h"
 #include "envoy/stats/scope.h"
 
 #include "common/common/logger.h"
@@ -21,6 +22,9 @@
 #include "common/signal/fatal_error_handler.h"
 
 namespace Envoy {
+namespace Network {
+class BufferedIoSocketHandleImpl;
+}
 namespace Event {
 
 /**
@@ -34,7 +38,9 @@ class DispatcherImpl : Logger::Loggable<Logger::Id::main>,
   DispatcherImpl(const std::string& name, Buffer::WatermarkFactoryPtr&& factory, Api::Api& api,
                  Event::TimeSystem& time_system);
   ~DispatcherImpl() override;
-
+  using InternalConnectionCallback =
+      std::function<void(const Network::Address::InstanceConstSharedPtr& address,
+                         std::unique_ptr<Network::ConnectionSocket> internal_socket)>;
   /**
    * @return event_base& the libevent base.
    */
@@ -53,6 +59,14 @@ class DispatcherImpl : Logger::Loggable<Logger::Id::main>,
                          Network::Address::InstanceConstSharedPtr source_address,
                          Network::TransportSocketPtr&& transport_socket,
                          const Network::ConnectionSocket::OptionsSharedPtr& options) override;
+  // Register the internal listener to setup the internal connection. Pass nullptr callback to
+  // unregister.
+  void registerInternalListener(const std::string& internal_listener_id,
+                                InternalConnectionCallback internal_conn_callback);
+
+  Network::ClientConnectionPtr
+  createInternalConnection(Network::Address::InstanceConstSharedPtr internal_address,
+                           Network::Address::InstanceConstSharedPtr local_address) override;
   Network::DnsResolverSharedPtr
   createDnsResolver(const std::vector<Network::Address::InstanceConstSharedPtr>& resolvers,
                     const bool use_tcp_for_dns_lookups) override;
@@ -121,6 +135,7 @@ class DispatcherImpl : Logger::Loggable<Logger::Id::main>,
   const ScopeTrackedObject* current_object_{};
   bool deferred_deleting_{};
   MonotonicTime approximate_monotonic_time_;
+  absl::flat_hash_map<std::string, InternalConnectionCallback> internal_listeners_;
 };
 
 } // namespace Event