SERVER-24162 Integrate TransportLayer

Expand the transport layer as needed to replace uses of abstract message port for ingress
networking.

1 files changed, 70 insertions, 9 deletions

diff --git a/src/mongo/transport/transport_layer.h b/src/mongo/transport/transport_layer.h
index 3028506c6f0..dde6248eaf5 100644
--- a/src/mongo/transport/transport_layer.h
+++ b/src/mongo/transport/transport_layer.h
@@ -57,7 +57,30 @@ class TransportLayer {
     MONGO_DISALLOW_COPYING(TransportLayer);
 
 public:
-    virtual ~TransportLayer();
+    /**
+     * Stats for sessions open in the Transport Layer.
+     */
+    struct Stats {
+        /**
+         * Returns the number of sessions currently open in the transport layer.
+         */
+        size_t numOpenSessions = 0;
+
+        /**
+         * Returns the total number of sessions that have ever been created by this TransportLayer.
+         */
+        size_t numCreatedSessions = 0;
+
+        /**
+         * Returns the number of available sessions we could still open. Only relevant
+         * when we are operating under a transport::Session limit (for example, in the
+         * legacy implementation, we respect a maximum number of connections). If there
+         * is no session limit, returns std::numeric_limits<int>::max().
+         */
+        size_t numAvailableSessions = 0;
+    };
+
+    virtual ~TransportLayer() = default;
 
     /**
      * Source (receive) a new Message for this Session.
@@ -101,7 +124,7 @@ public:
      * This thread may be used by the TransportLayer to run other Tickets that were
      * enqueued prior to this call.
      */
-    virtual Status wait(Ticket ticket) = 0;
+    virtual Status wait(Ticket&& ticket) = 0;
 
     /**
      * Callback for Tickets that are run via asyncWait().
@@ -115,7 +138,27 @@ public:
      * This thread will not be used by the TransportLayer to perform work. The callback
      * passed to asyncWait() may be run on any thread.
      */
-    virtual void asyncWait(Ticket ticket, TicketCallback callback) = 0;
+    virtual void asyncWait(Ticket&& ticket, TicketCallback callback) = 0;
+
+    /**
+     * Tag this Session within the TransportLayer with the tags currently assigned to the
+     * Session. If endAllSessions() is called with a matching
+     * Session::TagMask, this Session will not be ended.
+     *
+     * Before calling this method, use Session::replaceTags() to set the desired TagMask.
+     */
+    virtual void registerTags(const Session& session) = 0;
+
+    /**
+     * Return the stored X509 subject name for this session. If the session does not
+     * exist in this TransportLayer, returns "".
+     */
+    virtual std::string getX509SubjectName(const Session& session) = 0;
+
+    /**
+     * Returns the number of sessions currently open in the transport layer.
+     */
+    virtual Stats sessionStats() = 0;
 
     /**
      * End the given Session. Tickets for this Session that have already been
@@ -131,11 +174,20 @@ public:
     virtual void end(const Session& session) = 0;
 
     /**
-     * End all active sessions in the TransportLayer. Tickets that have already been
-     * started via wait() or asyncWait() will complete, but may return a failed Status.
-     * This method is synchronous and will not return until all sessions have ended.
+     * End all active sessions in the TransportLayer. Tickets that have already been started via
+     * wait() or asyncWait() will complete, but may return a failed Status.  This method is
+     * asynchronous and will return after all sessions have been notified to end.
+     *
+     * If a TagMask is provided, endAllSessions() will skip over sessions with matching
+     * tags and leave them open.
+     */
+    virtual void endAllSessions(Session::TagMask tags = Session::kEmptyTagMask) = 0;
+
+    /**
+     * Start the TransportLayer. After this point, the TransportLayer will begin accepting active
+     * sessions from new transport::Endpoints.
      */
-    virtual void endAllSessions() = 0;
+    virtual Status start() = 0;
 
     /**
      * Shut the TransportLayer down. After this point, the TransportLayer will
@@ -147,12 +199,21 @@ public:
     virtual void shutdown() = 0;
 
 protected:
-    TransportLayer();
+    TransportLayer() = default;
 
     /**
      * Return the implementation of this Ticket.
      */
-    TicketImpl* getTicketImpl(const Ticket& ticket);
+    TicketImpl* getTicketImpl(const Ticket& ticket) {
+        return ticket.impl();
+    }
+
+    /**
+     * Return the transport layer of this Ticket.
+     */
+    TransportLayer* getTicketTransportLayer(const Ticket& ticket) {
+        return ticket._tl;
+    }
 };
 
 }  // namespace transport