listener: extract active_tcp_socket and active_stream_listener_base

source/server/BUILD

@@ -64,9 +64,9 @@ envoy_cc_library(
 envoy_cc_library(
     name = "connection_handler_lib",
     deps = [
-        "//source/server:active_tcp_listener",
-        "//source/server:active_udp_listener",
-        "//source/server:connection_handler_impl",
+        ":active_tcp_listener",
+        ":active_udp_listener",
+        ":connection_handler_impl",
     ],
 )
 
@@ -116,25 +116,20 @@ envoy_cc_library(
         "active_tcp_listener.h",
     ],
     deps = [
+        ":active_stream_listener_base",
+        ":active_tcp_socket",
         "//envoy/common:time_interface",
         "//envoy/event:deferred_deletable",
         "//envoy/event:dispatcher_interface",
-        "//envoy/event:timer_interface",
         "//envoy/network:connection_handler_interface",
         "//envoy/network:connection_interface",
-        "//envoy/network:filter_interface",
         "//envoy/network:listen_socket_interface",
         "//envoy/network:listener_interface",
         "//envoy/server:listener_manager_interface",
-        "//envoy/stats:timespan_interface",
         "//source/common/common:assert_lib",
         "//source/common/common:linked_object",
-        "//source/common/common:non_copyable",
-        "//source/common/common:safe_memcpy_lib",
-        "//source/common/event:deferred_task",
         "//source/common/network:connection_lib",
         "//source/common/stats:timespan_lib",
-        "//source/common/stream_info:stream_info_lib",
         "//source/server:active_listener_base",
     ],
 )
@@ -157,6 +152,76 @@ envoy_cc_library(
     ],
 )
 
+envoy_cc_library(
+    name = "active_tcp_listener_headers",
+    hdrs = [
+        "active_stream_listener_base.h",
+        "active_tcp_socket.h",
+    ],
+    deps = [
+        ":active_listener_base",
+        "//envoy/common:time_interface",
+        "//envoy/event:deferred_deletable",
+        "//envoy/event:dispatcher_interface",
+        "//envoy/event:timer_interface",
+        "//envoy/network:connection_handler_interface",
+        "//envoy/network:connection_interface",
+        "//envoy/network:filter_interface",
+        "//envoy/network:listen_socket_interface",
+        "//envoy/network:listener_interface",
+        "//envoy/server:listener_manager_interface",
+        "//source/common/common:linked_object",
+    ],
+)
+
+envoy_cc_library(
+    name = "active_tcp_socket",
+    srcs = ["active_tcp_socket.cc"],
+    hdrs = [
+        "active_tcp_socket.h",
+    ],
+    deps = [
+        ":active_listener_base",
+        ":active_tcp_listener_headers",
+        "//envoy/common:time_interface",
+        "//envoy/event:deferred_deletable",
+        "//envoy/event:dispatcher_interface",
+        "//envoy/event:timer_interface",
+        "//envoy/network:connection_handler_interface",
+        "//envoy/network:connection_interface",
+        "//envoy/network:filter_interface",
+        "//envoy/network:listen_socket_interface",
+        "//envoy/network:listener_interface",
+        "//source/common/common:linked_object",
+        "//source/common/network:connection_lib",
+        "//source/common/stream_info:stream_info_lib",
+    ],
+)
+
+envoy_cc_library(
+    name = "active_stream_listener_base",
+    srcs = ["active_stream_listener_base.cc"],
+    hdrs = [
+        "active_stream_listener_base.h",
+    ],
+    deps = [
+        ":active_listener_base",
+        ":active_tcp_listener_headers",
+        "//envoy/common:time_interface",
+        "//envoy/event:deferred_deletable",
+        "//envoy/event:dispatcher_interface",
+        "//envoy/event:timer_interface",
+        "//envoy/network:connection_handler_interface",
+        "//envoy/network:connection_interface",
+        "//envoy/network:filter_interface",
+        "//envoy/network:listen_socket_interface",
+        "//envoy/network:listener_interface",
+        "//envoy/stream_info:stream_info_interface",
+        "//source/common/common:linked_object",
+        "//source/common/network:connection_lib",
+    ],
+)
+
 envoy_cc_library(
     name = "drain_manager_lib",
     srcs = ["drain_manager_impl.cc"],

source/server/active_stream_listener_base.cc

@@ -0,0 +1,62 @@
+#include "source/server/active_stream_listener_base.h"
+
+#include "envoy/network/filter.h"
+
+namespace Envoy {
+namespace Server {
+
+ActiveStreamListenerBase::ActiveStreamListenerBase(Network::ConnectionHandler& parent,
+                                                   Event::Dispatcher& dispatcher,
+                                                   Network::ListenerPtr&& listener,
+                                                   Network::ListenerConfig& config)
+    : ActiveListenerImplBase(parent, &config), parent_(parent),
+      listener_filters_timeout_(config.listenerFiltersTimeout()),
+      continue_on_listener_filters_timeout_(config.continueOnListenerFiltersTimeout()),
+      dispatcher_(dispatcher), listener_(std::move(listener)) {}
+
+void ActiveStreamListenerBase::emitLogs(Network::ListenerConfig& config,
+                                        StreamInfo::StreamInfo& stream_info) {
+  stream_info.onRequestComplete();
+  for (const auto& access_log : config.accessLogs()) {
+    access_log->log(nullptr, nullptr, nullptr, stream_info);
+  }
+}
+
+void ActiveStreamListenerBase::newConnection(Network::ConnectionSocketPtr&& socket,
+                                             std::unique_ptr<StreamInfo::StreamInfo> stream_info) {
+  // Find matching filter chain.
+  const auto filter_chain = config_->filterChainManager().findFilterChain(*socket);
+  if (filter_chain == nullptr) {
+    RELEASE_ASSERT(socket->addressProvider().remoteAddress() != nullptr, "");
+    ENVOY_LOG(debug, "closing connection from {}: no matching filter chain found",
+              socket->addressProvider().remoteAddress()->asString());
+    stats_.no_filter_chain_match_.inc();
+    stream_info->setResponseFlag(StreamInfo::ResponseFlag::NoRouteFound);
+    stream_info->setResponseCodeDetails(StreamInfo::ResponseCodeDetails::get().FilterChainNotFound);
+    emitLogs(*config_, *stream_info);
+    socket->close();
+    return;
+  }
+  stream_info->setFilterChainName(filter_chain->name());
+  auto transport_socket = filter_chain->transportSocketFactory().createTransportSocket(nullptr);
+  stream_info->setDownstreamSslConnection(transport_socket->ssl());
+  auto server_conn_ptr = dispatcher().createServerConnection(
+      std::move(socket), std::move(transport_socket), *stream_info);
+  if (const auto timeout = filter_chain->transportSocketConnectTimeout();
+      timeout != std::chrono::milliseconds::zero()) {
+    server_conn_ptr->setTransportSocketConnectTimeout(timeout);
+  }
+  server_conn_ptr->setBufferLimits(config_->perConnectionBufferLimitBytes());
+  RELEASE_ASSERT(server_conn_ptr->addressProvider().remoteAddress() != nullptr, "");
+  const bool empty_filter_chain = !config_->filterChainFactory().createNetworkFilterChain(
+      *server_conn_ptr, filter_chain->networkFilterFactories());
+  if (empty_filter_chain) {
+    ENVOY_CONN_LOG(debug, "closing connection from {}: no filters", *server_conn_ptr,
+                   server_conn_ptr->addressProvider().remoteAddress()->asString());
+    server_conn_ptr->close(Network::ConnectionCloseType::NoFlush);
+  }
+  newActiveConnection(*filter_chain, std::move(server_conn_ptr), std::move(stream_info));
+}
+
+} // namespace Server
+} // namespace Envoy

source/server/active_stream_listener_base.h

@@ -0,0 +1,115 @@
+#pragma once
+
+#include <atomic>
+#include <cstdint>
+#include <list>
+#include <memory>
+
+#include "envoy/common/time.h"
+#include "envoy/event/dispatcher.h"
+#include "envoy/network/connection.h"
+#include "envoy/network/connection_handler.h"
+#include "envoy/network/listener.h"
+#include "envoy/stream_info/stream_info.h"
+
+#include "source/common/common/linked_object.h"
+#include "source/server/active_listener_base.h"
+#include "source/server/active_tcp_socket.h"
+
+namespace Envoy {
+namespace Server {
+
+// The base class of the stream listener. It owns the the active sockets that drive itself through
+// the listener filters. After the active socket passes all the listener filters, a server
+// connection is created. The derived listener must override ``newActiveConnection`` to take the
+// ownership of that server connection.
+class ActiveStreamListenerBase : public ActiveListenerImplBase,
+                                 protected Logger::Loggable<Logger::Id::conn_handler> {
+public:
+  ActiveStreamListenerBase(Network::ConnectionHandler& parent, Event::Dispatcher& dispatcher,
+                           Network::ListenerPtr&& listener, Network::ListenerConfig& config);
+  static void emitLogs(Network::ListenerConfig& config, StreamInfo::StreamInfo& stream_info);
+
+  Event::Dispatcher& dispatcher() { return dispatcher_; }
+
+  /**
+   * Schedule to remove and destroy the active connections which are not tracked by listener
+   * config. Caution: The connection are not destroyed yet when function returns.
+   */
+  void
+  deferredRemoveFilterChains(const std::list<const Network::FilterChain*>& draining_filter_chains) {
+    // Need to recover the original deleting state.
+    const bool was_deleting = is_deleting_;
+    is_deleting_ = true;
+    for (const auto* filter_chain : draining_filter_chains) {
+      deferRemoveFilterChain(filter_chain);
+    }
+    is_deleting_ = was_deleting;
+  }
+
+  virtual void incNumConnections() PURE;
+  virtual void decNumConnections() PURE;
+
+  /**
+   * Create a new connection from a socket accepted by the listener.
+   */
+  void newConnection(Network::ConnectionSocketPtr&& socket,
+                     std::unique_ptr<StreamInfo::StreamInfo> stream_info);
+  /**
+   * Schedule to remove and destroy the active connections owned by the filter chain.
+   */
+  virtual void deferRemoveFilterChain(const Network::FilterChain* filter_chain) PURE;
+
+  virtual Network::BalancedConnectionHandlerOptRef
+  getBalancedHandlerByAddress(const Network::Address::Instance& address) PURE;
+
+  void onSocketAccepted(std::unique_ptr<ActiveTcpSocket> active_socket) {
+    // Create and run the filters
+    config_->filterChainFactory().createListenerFilterChain(*active_socket);
+    active_socket->continueFilterChain(true);
+
+    // Move active_socket to the sockets_ list if filter iteration needs to continue later.
+    // Otherwise we let active_socket be destructed when it goes out of scope.
+    if (active_socket->iter_ != active_socket->accept_filters_.end()) {
+      active_socket->startTimer();
+      LinkedList::moveIntoListBack(std::move(active_socket), sockets_);
+    } else {
+      if (!active_socket->connected_) {
+        // If active_socket is about to be destructed, emit logs if a connection is not created.
+        if (active_socket->stream_info_ != nullptr) {
+          emitLogs(*config_, *active_socket->stream_info_);
+        } else {
+          // If the active_socket is not connected, this socket is not promoted to active
+          // connection. Thus the stream_info_ is owned by this active socket.
+          ENVOY_BUG(active_socket->stream_info_ != nullptr,
+                    "the unconnected active socket must have stream info.");
+        }
+      }
+    }
+  }
+
+  // Below members are open to access by ActiveTcpSocket.
+  Network::ConnectionHandler& parent_;
+  const std::chrono::milliseconds listener_filters_timeout_;
+  const bool continue_on_listener_filters_timeout_;
+  std::list<std::unique_ptr<ActiveTcpSocket>> sockets_;
+
+protected:
+  /**
+   * Create the active connection from server connection. This active listener owns the created
+   * active connection.
+   *
+   * @param filter_chain The network filter chain linking to the connection.
+   * @param server_conn_ptr The server connection.
+   * @param stream_info The stream info of the active connection.
+   */
+  virtual void newActiveConnection(const Network::FilterChain& filter_chain,
+                                   Network::ServerConnectionPtr server_conn_ptr,
+                                   std::unique_ptr<StreamInfo::StreamInfo> stream_info) PURE;
+  Event::Dispatcher& dispatcher_;
+  Network::ListenerPtr listener_;
+  bool is_deleting_{false};
+};
+
+} // namespace Server
+} // namespace Envoy