Enable setup of ConnectionEventCallback on RemoteAcceptor

Summary:
# Issue
Currently the `ConnectionEventCallback::onConnectionDequeuedByAcceptorCallback()` method is never called on `ThriftServer`, as `RemoteAcceptor` object is always initialized with `nullprt`. This blocks our ability to implement a Thrift counter which compares enqued and dequeued events to calculate current inflight connection count waiting in the queue.

The reason of this issue is because on [ThriftServer::setup()](https://fburl.com/code/n3dyjy0t),  we bind the port, which creates `AsyncServerSocket` and related `RemoteAcceptor` in the moment when `ConnectionEventCallback` is not initialized yet. The `ConnectionEventCallback` would only be set later on the `AsyncServerSocket` during setup process ([see here](https://fburl.com/code/bryagy6k)).

There are no `dequeue` events in `thrift_connections_event` scuba, which makes me believe that logging those on the `RemoteAcceptor` never worked:
https://fburl.com/scuba/thrift_connection_events/vu2kjvh5

# Fix
Add ability to set `ConnectionEventCallback` in the `ServerBootstrap` class in Wangle and use that callback during address `bind()` process.

Reviewed By: robertroeser

Differential Revision: D64274597

fbshipit-source-id: fd9b7e0635b4f46678109c5b4b1786b163d8ab3b

wangle/bootstrap/ServerBootstrap.h

@@ -79,6 +79,13 @@ class ServerBootstrap {
     return this;
   }
 
+  ServerBootstrap* useConnectionEventCallback(
+      std::shared_ptr<folly::AsyncServerSocket::ConnectionEventCallback>
+          connectionEventCallback) {
+    connectionEventCallback_ = connectionEventCallback;
+    return this;
+  }
+
   /*
    * BACKWARDS COMPATIBILITY - an acceptor factory can be set.  Your
    * Acceptor is responsible for managing the connection pool.
@@ -234,7 +241,11 @@ class ServerBootstrap {
     auto startupFunc = [&](std::shared_ptr<folly::Baton<>> barrier) {
       try {
         auto socket = socketFactory_->newSocket(
-            address, socketConfig.acceptBacklog, reusePort, socketConfig);
+            address,
+            socketConfig.acceptBacklog,
+            reusePort,
+            socketConfig,
+            connectionEventCallback_.get());
         sock_lock.lock();
         new_sockets.push_back(socket);
         sock_lock.unlock();
@@ -385,6 +396,8 @@ class ServerBootstrap {
       std::make_unique<folly::Baton<>>()};
   bool stopped_{false};
   bool useSharedSSLContextManager_{false};
+  std::shared_ptr<folly::AsyncServerSocket::ConnectionEventCallback>
+      connectionEventCallback_;
 };
 
 } // namespace wangle

wangle/bootstrap/ServerSocketFactory.h

@@ -29,7 +29,9 @@ class ServerSocketFactory {
       folly::SocketAddress address,
       int backlog,
       bool reuse,
-      const ServerSocketConfig& config) = 0;
+      const ServerSocketConfig& config,
+      folly::AsyncServerSocket::ConnectionEventCallback*
+          connectionEventCallback) = 0;
 
   virtual void removeAcceptCB(
       std::shared_ptr<folly::AsyncSocketBase> sock,
@@ -50,13 +52,18 @@ class AsyncServerSocketFactory : public ServerSocketFactory {
       folly::SocketAddress address,
       int /*backlog*/,
       bool reuse,
-      const ServerSocketConfig& config) override {
+      const ServerSocketConfig& config,
+      folly::AsyncServerSocket::ConnectionEventCallback*
+          connectionEventCallback) override {
     auto* evb = folly::EventBaseManager::get()->getEventBase();
     std::shared_ptr<folly::AsyncServerSocket> socket(
         new folly::AsyncServerSocket(evb), ThreadSafeDestructor());
     if (config.useZeroCopy) {
       socket->setZeroCopy(true);
     }
+    if (connectionEventCallback != nullptr) {
+      socket->setConnectionEventCallback(connectionEventCallback);
+    }
     socket->setMaxNumMessagesInQueue(config.maxNumPendingConnectionsPerWorker);
     socket->setReusePortEnabled(reuse);
     if (config.enableTCPFastOpen) {
@@ -108,7 +115,8 @@ class AsyncUDPServerSocketFactory : public ServerSocketFactory {
       folly::SocketAddress address,
       int /*backlog*/,
       bool reuse,
-      const ServerSocketConfig& /*config*/) override {
+      const ServerSocketConfig& /*config*/,
+      folly::AsyncServerSocket::ConnectionEventCallback*) override {
     folly::EventBase* evb = folly::EventBaseManager::get()->getEventBase();
     std::shared_ptr<folly::AsyncUDPServerSocket> socket(
         new folly::AsyncUDPServerSocket(evb), ThreadSafeDestructor());