SERVER-30992 Integrate mongobridge with ServiceEntryPoint

2 files changed, 288 insertions, 277 deletions

diff --git a/src/mongo/tools/SConscript b/src/mongo/tools/SConscript
index 97392ef4963..7c4ce83c43b 100644
--- a/src/mongo/tools/SConscript
+++ b/src/mongo/tools/SConscript
@@ -14,9 +14,14 @@ mongobridge = env.Program(
         "mongobridge_options_init.cpp"
     ],
     LIBDEPS=[
-        "$BUILD_DIR/mongo/client/clientdriver",
-        "$BUILD_DIR/mongo/util/signal_handlers",
-        "$BUILD_DIR/mongo/util/options_parser/options_parser_init",
+        '$BUILD_DIR/mongo/db/dbmessage',
+        '$BUILD_DIR/mongo/rpc/rpc',
+        '$BUILD_DIR/mongo/transport/service_entry_point',
+        '$BUILD_DIR/mongo/transport/service_executor',
+        '$BUILD_DIR/mongo/transport/transport_layer',
+        '$BUILD_DIR/mongo/util/net/network',
+        '$BUILD_DIR/mongo/util/options_parser/options_parser_init',
+        '$BUILD_DIR/mongo/util/signal_handlers',
     ],
     INSTALL_ALIAS=[
         'tools'
diff --git a/src/mongo/tools/bridge.cpp b/src/mongo/tools/bridge.cpp
index 5e5fcbf7417..7d4a1f10e8c 100644
--- a/src/mongo/tools/bridge.cpp
+++ b/src/mongo/tools/bridge.cpp
@@ -35,8 +35,8 @@
 
 #include "mongo/base/init.h"
 #include "mongo/base/initializer.h"
-#include "mongo/client/dbclientinterface.h"
 #include "mongo/db/dbmessage.h"
+#include "mongo/db/operation_context.h"
 #include "mongo/db/service_context.h"
 #include "mongo/db/service_context_noop.h"
 #include "mongo/db/service_context_registrar.h"
@@ -50,13 +50,14 @@
 #include "mongo/stdx/thread.h"
 #include "mongo/tools/bridge_commands.h"
 #include "mongo/tools/mongobridge_options.h"
+#include "mongo/transport/message_compressor_manager.h"
+#include "mongo/transport/service_entry_point_impl.h"
+#include "mongo/transport/service_executor_synchronous.h"
 #include "mongo/transport/transport_layer_asio.h"
 #include "mongo/util/assert_util.h"
 #include "mongo/util/exit.h"
 #include "mongo/util/log.h"
 #include "mongo/util/mongoutils/str.h"
-#include "mongo/util/net/abstract_message_port.h"
-#include "mongo/util/net/listen.h"
 #include "mongo/util/net/message.h"
 #include "mongo/util/quick_exit.h"
 #include "mongo/util/signal_handlers.h"
@@ -84,237 +85,24 @@ boost::optional<HostAndPort> extractHostInfo(const OpMsgRequest& request) {
     return boost::none;
 }
 
-class Forwarder {
-public:
-    Forwarder(AbstractMessagingPort* mp,
-              stdx::mutex* settingsMutex,
-              HostSettingsMap* settings,
-              int64_t seed)
-        : _mp(mp), _settingsMutex(settingsMutex), _settings(settings), _prng(seed) {}
-
-    void operator()() {
-        transport::SessionHandle dest = []() -> transport::SessionHandle {
-            HostAndPort destAddr{mongoBridgeGlobalParams.destUri};
-            const Seconds kConnectTimeout(30);
-            auto now = getGlobalServiceContext()->getFastClockSource()->now();
-            const auto connectExpiration = now + kConnectTimeout;
-            while (now < connectExpiration) {
-                auto tl = getGlobalServiceContext()->getTransportLayer();
-                auto sws =
-                    tl->connect(destAddr, transport::kGlobalSSLMode, connectExpiration - now);
-                auto status = sws.getStatus();
-                if (!status.isOK()) {
-                    warning() << "Unable to establish connection to " << destAddr << ": " << status;
-                    now = getGlobalServiceContext()->getFastClockSource()->now();
-                } else {
-                    return std::move(sws.getValue());
-                }
-
-                sleepmillis(500);
-            }
-
-            return nullptr;
-        }();
-
-        if (!dest) {
-            log() << "end connection " << _mp->remote();
-            _mp->shutdown();
-            return;
-        }
-
-        bool receivingFirstMessage = true;
-        boost::optional<HostAndPort> host;
-
-        Message request;
-        Message response;
-        MessageCompressorManager compressorManager;
-
-        while (true) {
-            try {
-                request.reset();
-                if (!_mp->recv(request)) {
-                    log() << "end connection " << _mp->remote().toString();
-                    _mp->shutdown();
-                    break;
-                }
-
-                uassert(ErrorCodes::IllegalOperation,
-                        str::stream() << "Unsupported network op " << request.operation(),
-                        isSupportedRequestNetworkOp(request.operation()));
-
-                if (request.operation() == dbCompressed) {
-                    auto swm = compressorManager.decompressMessage(request);
-                    if (!swm.isOK()) {
-                        error() << "Error decompressing message: " << swm.getStatus();
-                        _mp->shutdown();
-                        return;
-                    }
-                    request = std::move(swm.getValue());
-                }
-
-                const bool isFireAndForgetCommand = OpMsg::isFlagSet(request, OpMsg::kMoreToCome);
-
-                boost::optional<OpMsgRequest> cmdRequest;
-                if ((request.operation() == dbQuery &&
-                     NamespaceString(DbMessage(request).getns()).isCommand()) ||
-                    request.operation() == dbCommand || request.operation() == dbMsg) {
-                    cmdRequest = rpc::opMsgRequestFromAnyProtocol(request);
-                    if (receivingFirstMessage) {
-                        host = extractHostInfo(*cmdRequest);
-                    }
-
-                    std::string hostName = host ? (host->toString()) : "<unknown>";
-                    LOG(1) << "Received \"" << cmdRequest->getCommandName()
-                           << "\" command with arguments " << cmdRequest->body << " from "
-                           << hostName;
-                }
-                receivingFirstMessage = false;
-
-                // Handle a message intended to configure the mongobridge and return a response.
-                // The 'request' is consumed by the mongobridge and does not get forwarded to
-                // 'dest'.
-                if (auto status = maybeProcessBridgeCommand(cmdRequest)) {
-                    invariant(!isFireAndForgetCommand);
-
-                    auto replyBuilder = rpc::makeReplyBuilder(rpc::protocolForMessage(request));
-                    BSONObj metadata;
-                    BSONObj reply;
-                    StatusWith<BSONObj> commandReply(reply);
-                    if (!status->isOK()) {
-                        commandReply = StatusWith<BSONObj>(*status);
-                    }
-                    auto cmdResponse = replyBuilder->setCommandReply(std::move(commandReply))
-                                           .setMetadata(metadata)
-                                           .done();
-                    cmdResponse.header().setId(nextMessageId());
-                    cmdResponse.header().setResponseToMsgId(request.header().getId());
-                    _mp->say(cmdResponse);
-                    continue;
-                }
-
-                // Get the message handling settings for 'host' if the source of _mp's connection is
-                // known. By default, messages are forwarded to 'dest' without any additional delay.
-                HostSettings hostSettings = getHostSettings(host);
-
-                switch (hostSettings.state) {
-                    // Forward the message to 'dest' after waiting for 'hostSettings.delay'
-                    // milliseconds.
-                    case HostSettings::State::kForward:
-                        sleepmillis(durationCount<Milliseconds>(hostSettings.delay));
-                        break;
-                    // Close the connection to 'dest'.
-                    case HostSettings::State::kHangUp:
-                        log() << "Rejecting connection from " << host->toString()
-                              << ", end connection " << _mp->remote().toString();
-                        _mp->shutdown();
-                        return;
-                    // Forward the message to 'dest' with probability '1 - hostSettings.loss'.
-                    case HostSettings::State::kDiscard:
-                        if (_prng.nextCanonicalDouble() < hostSettings.loss) {
-                            std::string hostName = host ? (host->toString()) : "<unknown>";
-                            if (cmdRequest) {
-                                log() << "Discarding \"" << cmdRequest->getCommandName()
-                                      << "\" command with arguments " << cmdRequest->body
-                                      << " from " << hostName;
-                            } else {
-                                log() << "Discarding " << networkOpToString(request.operation())
-                                      << " from " << hostName;
-                            }
-                            continue;
-                        }
-                        break;
-                }
+ServiceContextRegistrar serviceContextCreator([]() {
+    return std::make_unique<ServiceContextNoop>();
+});
 
-                // Send the message we received from '_mp' to 'dest'. 'dest' returns a response for
-                // OP_QUERY, OP_GET_MORE, and OP_COMMAND messages that we respond back to
-                // '_mp' with.
-                if (!isFireAndForgetCommand &&
-                    (request.operation() == dbQuery || request.operation() == dbGetMore ||
-                     request.operation() == dbCommand || request.operation() == dbMsg)) {
-                    // TODO dbMsg moreToCome
-                    // Forward the message to 'dest' and receive its reply in 'response'.
-                    uassertStatusOK(dest->sinkMessage(request));
-                    response = uassertStatusOK(dest->sourceMessage());
-                    uassert(50727,
-                            "Response ID did not match the sent message ID.",
-                            response.header().getResponseToMsgId() == request.header().getId());
-
-                    // If there's nothing to respond back to '_mp' with, then close the connection.
-                    if (response.empty()) {
-                        log() << "Received an empty response, end connection "
-                              << _mp->remote().toString();
-                        _mp->shutdown();
-                        break;
-                    }
-
-                    // Reload the message handling settings for 'host' in case they were changed
-                    // while waiting for a response from 'dest'.
-                    hostSettings = getHostSettings(host);
-
-                    // It's possible that sending 'request' blocked until 'dest' had something to
-                    // reply with. If the message handling settings were since changed to close
-                    // connections from 'host', then do so now.
-                    if (hostSettings.state == HostSettings::State::kHangUp) {
-                        log() << "Closing connection from " << host->toString()
-                              << ", end connection " << _mp->remote().toString();
-                        _mp->shutdown();
-                        break;
-                    }
-
-                    _mp->say(response);
-
-                    // If 'exhaust' is true, then instead of trying to receive another message from
-                    // '_mp', receive messages from 'dest' until it returns a cursor id of zero.
-                    bool exhaust = false;
-                    if (request.operation() == dbQuery) {
-                        DbMessage d(request);
-                        QueryMessage q(d);
-                        exhaust = q.queryOptions & QueryOption_Exhaust;
-                    }
-                    while (exhaust) {
-                        if (response.operation() == dbCompressed) {
-                            auto swm = compressorManager.decompressMessage(response);
-                            if (!swm.isOK()) {
-                                error() << "Error decompressing message: " << swm.getStatus();
-                                _mp->shutdown();
-                                return;
-                            }
-                            response = std::move(swm.getValue());
-                        }
-
-                        MsgData::View header = response.header();
-                        QueryResult::View qr = header.view2ptr();
-                        if (qr.getCursorId()) {
-                            response = uassertStatusOK(dest->sourceMessage());
-                            _mp->say(response);
-                        } else {
-                            exhaust = false;
-                        }
-                    }
-                } else {
-                    uassertStatusOK(dest->sinkMessage(request));
-                }
-            } catch (const DBException& ex) {
-                error() << "Caught DBException in Forwarder: " << ex << ", end connection "
-                        << _mp->remote().toString();
-                _mp->shutdown();
-                break;
-            } catch (...) {
-                severe() << exceptionToStatus() << ", terminating";
-                quickExit(EXIT_UNCAUGHT);
-            }
-        }
-    }
+}  // namespace
 
-private:
+class BridgeContext {
+public:
     Status runBridgeCommand(StringData cmdName, BSONObj cmdObj) {
         auto status = BridgeCommand::findCommand(cmdName);
         if (!status.isOK()) {
             return status.getStatus();
         }
 
+        log() << "Processing bridge command: " << cmdName;
+
         BridgeCommand* command = status.getValue();
-        return command->run(cmdObj, _settingsMutex, _settings);
+        return command->run(cmdObj, &_settingsMutex, &_settings);
     }
 
     boost::optional<Status> maybeProcessBridgeCommand(boost::optional<OpMsgRequest> cmdRequest) {
@@ -334,69 +122,277 @@ private:
 
     HostSettings getHostSettings(boost::optional<HostAndPort> host) {
         if (host) {
-            stdx::lock_guard<stdx::mutex> lk(*_settingsMutex);
-            return (*_settings)[*host];
+            stdx::lock_guard<stdx::mutex> lk(_settingsMutex);
+            return (_settings)[*host];
         }
         return {};
     }
 
-    AbstractMessagingPort* _mp;
+    PseudoRandom makeSeededPRNG() {
+        static PseudoRandom globalPRNG(mongoBridgeGlobalParams.seed);
+        return PseudoRandom(globalPRNG.nextInt64());
+    }
 
-    stdx::mutex* _settingsMutex;
-    HostSettingsMap* _settings;
+    static BridgeContext* get();
 
-    PseudoRandom _prng;
+private:
+    static const ServiceContext::Decoration<BridgeContext> _get;
+
+    stdx::mutex _settingsMutex;
+    HostSettingsMap _settings;
 };
 
-class BridgeListener final : public Listener {
+const ServiceContextNoop::Decoration<BridgeContext> BridgeContext::_get =
+    ServiceContext::declareDecoration<BridgeContext>();
+
+BridgeContext* BridgeContext::get() {
+    return &_get(getGlobalServiceContext());
+}
+
+class ServiceEntryPointBridge;
+class ProxiedConnection {
 public:
-    BridgeListener()
-        : Listener(
-              "bridge", "0.0.0.0", mongoBridgeGlobalParams.port, getGlobalServiceContext(), false),
-          _seedSource(mongoBridgeGlobalParams.seed) {
-        log() << "Setting random seed: " << mongoBridgeGlobalParams.seed;
+    ProxiedConnection() : _dest(nullptr), _prng(BridgeContext::get()->makeSeededPRNG()) {}
+
+    transport::Session* operator->() {
+        return _dest.get();
     }
 
-    void accepted(std::unique_ptr<AbstractMessagingPort> mp) override final {
-        {
-            stdx::lock_guard<stdx::mutex> lk(_portsMutex);
-            if (_inShutdown.load()) {
-                mp->shutdown();
-                return;
-            }
-            _ports.insert(mp.get());
+    const boost::optional<HostAndPort>& host() const {
+        return _host;
+    }
+
+    std::string toString() const {
+        if (_host) {
+            return _host->toString();
         }
+        return "<unknown>";
+    }
 
-        Forwarder f(mp.release(), &_settingsMutex, &_settings, _seedSource.nextInt64());
-        stdx::thread t(f);
-        t.detach();
+    void setExhaust(bool val) {
+        _inExhaust = val;
     }
 
-    void shutdownAll() {
-        stdx::lock_guard<stdx::mutex> lk(_portsMutex);
-        for (auto mp : _ports) {
-            mp->shutdown();
+    bool inExhaust() const {
+        return _inExhaust;
+    }
+
+    void extractHostInfo(OpMsgRequest request) {
+        if (_seenFirstMessage)
+            return;
+        _seenFirstMessage = true;
+
+        // The initial isMaster request made by mongod and mongos processes should contain a
+        // hostInfo field that identifies the process by its host:port.
+        StringData cmdName = request.getCommandName();
+        if (cmdName != "isMaster" && cmdName != "ismaster") {
+            return;
         }
+
+        if (auto hostInfoElem = request.body["hostInfo"]) {
+            if (hostInfoElem.type() == String) {
+                _host = HostAndPort{hostInfoElem.valueStringData()};
+            }
+        }
+    }
+
+    double nextCanonicalDouble() {
+        return _prng.nextCanonicalDouble();
     }
 
+    static ProxiedConnection& get(const transport::SessionHandle& session);
+
 private:
-    stdx::mutex _portsMutex;
-    std::set<AbstractMessagingPort*> _ports;
-    AtomicWord<bool> _inShutdown{false};
+    friend class ServiceEntryPointBridge;
 
-    stdx::mutex _settingsMutex;
-    HostSettingsMap _settings;
+    static const transport::Session::Decoration<ProxiedConnection> _get;
+    transport::SessionHandle _dest;
+    PseudoRandom _prng;
+    boost::optional<HostAndPort> _host;
+    bool _seenFirstMessage = false;
+    bool _inExhaust = false;
+};
+
+const transport::Session::Decoration<ProxiedConnection> ProxiedConnection::_get =
+    transport::Session::declareDecoration<ProxiedConnection>();
+
+ProxiedConnection& ProxiedConnection::get(const transport::SessionHandle& session) {
+    return _get(*session);
+}
+
+class ServiceEntryPointBridge final : public ServiceEntryPointImpl {
+public:
+    explicit ServiceEntryPointBridge(ServiceContext* svcCtx) : ServiceEntryPointImpl(svcCtx) {}
 
-    PseudoRandom _seedSource;
+    void startSession(transport::SessionHandle session) final;
+    DbResponse handleRequest(OperationContext* opCtx, const Message& request) final;
 };
 
-std::unique_ptr<mongo::BridgeListener> listener;
+void ServiceEntryPointBridge::startSession(transport::SessionHandle session) {
+    transport::SessionHandle dest = []() -> transport::SessionHandle {
+        HostAndPort destAddr{mongoBridgeGlobalParams.destUri};
+        const Seconds kConnectTimeout(30);
+        auto now = getGlobalServiceContext()->getFastClockSource()->now();
+        const auto connectExpiration = now + kConnectTimeout;
+        while (now < connectExpiration) {
+            auto tl = getGlobalServiceContext()->getTransportLayer();
+            auto sws = tl->connect(destAddr, transport::kGlobalSSLMode, connectExpiration - now);
+            auto status = sws.getStatus();
+            if (!status.isOK()) {
+                warning() << "Unable to establish connection to " << destAddr << ": " << status;
+                now = getGlobalServiceContext()->getFastClockSource()->now();
+            } else {
+                return std::move(sws.getValue());
+            }
 
-ServiceContextRegistrar serviceContextCreator([]() {
-    return std::make_unique<ServiceContextNoop>();
-});
+            sleepmillis(500);
+        }
 
-}  // namespace
+        return nullptr;
+    }();
+
+    if (!dest) {
+        log() << "end connection " << session->remote();
+        return;
+    }
+
+    auto& proxiedConn = ProxiedConnection::get(session);
+    proxiedConn._dest = std::move(dest);
+
+    ServiceEntryPointImpl::startSession(std::move(session));
+}
+
+DbResponse ServiceEntryPointBridge::handleRequest(OperationContext* opCtx, const Message& request) {
+    const auto& source = opCtx->getClient()->session();
+    auto& dest = ProxiedConnection::get(source);
+    auto brCtx = BridgeContext::get();
+
+    if (dest.inExhaust()) {
+        DbMessage dbm(request);
+
+        auto response = uassertStatusOK(dest->sourceMessage());
+        if (response.operation() == dbCompressed) {
+            MessageCompressorManager compressorMgr;
+            response = uassertStatusOK(compressorMgr.decompressMessage(response));
+        }
+
+        MsgData::View header = response.header();
+        QueryResult::View qr = header.view2ptr();
+        if (qr.getCursorId()) {
+            return {std::move(response)};
+        } else {
+            dest.setExhaust(false);
+            return {Message(), dbm.getns()};
+        }
+    }
+
+    const bool isFireAndForgetCommand = OpMsg::isFlagSet(request, OpMsg::kMoreToCome);
+
+    boost::optional<OpMsgRequest> cmdRequest;
+    if ((request.operation() == dbQuery &&
+         NamespaceString(DbMessage(request).getns()).isCommand()) ||
+        request.operation() == dbCommand || request.operation() == dbMsg) {
+        cmdRequest = rpc::opMsgRequestFromAnyProtocol(request);
+
+        dest.extractHostInfo(*cmdRequest);
+
+        LOG(1) << "Received \"" << cmdRequest->getCommandName() << "\" command with arguments "
+               << cmdRequest->body << " from " << dest;
+    }
+
+    // Handle a message intended to configure the mongobridge and return a response.
+    // The 'request' is consumed by the mongobridge and does not get forwarded to
+    // 'dest'.
+    if (auto status = brCtx->maybeProcessBridgeCommand(cmdRequest)) {
+        invariant(!isFireAndForgetCommand);
+
+        auto replyBuilder = rpc::makeReplyBuilder(rpc::protocolForMessage(request));
+        BSONObj metadata;
+        BSONObj reply;
+        StatusWith<BSONObj> commandReply(reply);
+        if (!status->isOK()) {
+            commandReply = StatusWith<BSONObj>(*status);
+        }
+        return {
+            replyBuilder->setCommandReply(std::move(commandReply)).setMetadata(metadata).done()};
+    }
+
+
+    // Get the message handling settings for 'host' if the source of the connection is
+    // known. By default, messages are forwarded to 'dest' without any additional delay.
+    HostSettings hostSettings = brCtx->getHostSettings(dest.host());
+
+    switch (hostSettings.state) {
+        // Close the connection to 'dest'.
+        case HostSettings::State::kHangUp:
+            log() << "Rejecting connection from " << dest << ", end connection "
+                  << source->remote().toString();
+            source->end();
+            return {Message()};
+        // Forward the message to 'dest' with probability '1 - hostSettings.loss'.
+        case HostSettings::State::kDiscard:
+            if (dest.nextCanonicalDouble() < hostSettings.loss) {
+                std::string hostName = dest.toString();
+                if (cmdRequest) {
+                    log() << "Discarding \"" << cmdRequest->getCommandName()
+                          << "\" command with arguments " << cmdRequest->body << " from "
+                          << hostName;
+                } else {
+                    log() << "Discarding " << networkOpToString(request.operation()) << " from "
+                          << hostName;
+                }
+                return {Message()};
+            }
+        // Forward the message to 'dest' after waiting for 'hostSettings.delay'
+        // milliseconds.
+        case HostSettings::State::kForward:
+            sleepmillis(durationCount<Milliseconds>(hostSettings.delay));
+            break;
+    }
+
+    uassertStatusOK(dest->sinkMessage(request));
+
+    // Send the message we received from 'source' to 'dest'. 'dest' returns a response for
+    // OP_QUERY, OP_GET_MORE, and OP_COMMAND messages that we respond with.
+    if (!isFireAndForgetCommand &&
+        (request.operation() == dbQuery || request.operation() == dbGetMore ||
+         request.operation() == dbCommand || request.operation() == dbMsg)) {
+        // TODO dbMsg moreToCome
+        // Forward the message to 'dest' and receive its reply in 'response'.
+        auto response = uassertStatusOK(dest->sourceMessage());
+        uassert(50765,
+                "Response ID did not match the sent message ID.",
+                response.header().getResponseToMsgId() == request.header().getId());
+
+        // Reload the message handling settings for 'host' in case they were changed
+        // while waiting for a response from 'dest'.
+        hostSettings = brCtx->getHostSettings(dest.host());
+
+        // It's possible that sending 'request' blocked until 'dest' had something to
+        // reply with. If the message handling settings were since changed to close
+        // connections from 'host', then do so now.
+        if (hostSettings.state == HostSettings::State::kHangUp) {
+            log() << "Closing connection from " << dest << ", end connection " << source->remote();
+            source->end();
+            return {Message()};
+        }
+
+        std::string exhaustNS;
+        if (request.operation() == dbQuery) {
+            DbMessage d(request);
+            QueryMessage q(d);
+            dest.setExhaust(q.queryOptions & QueryOption_Exhaust);
+            if (dest.inExhaust()) {
+                exhaustNS = d.getns();
+            }
+        } else {
+            dest.setExhaust(false);
+        }
+        return {std::move(response), exhaustNS};
+    } else {
+        return {Message()};
+    }
+}
 
 int bridgeMain(int argc, char** argv, char** envp) {
 
@@ -404,8 +400,16 @@ int bridgeMain(int argc, char** argv, char** envp) {
         // NOTE: This function may be called at any time. It must not
         // depend on the prior execution of mongo initializers or the
         // existence of threads.
-        ListeningSockets::get()->closeAll();
-        listener->shutdownAll();
+        if (hasGlobalServiceContext()) {
+            auto sc = getGlobalServiceContext();
+            if (sc->getTransportLayer())
+                sc->getTransportLayer()->shutdown();
+
+            if (sc->getServiceEntryPoint()) {
+                sc->getServiceEntryPoint()->endAllSessions(transport::Session::kEmptyTagMask);
+                sc->getServiceEntryPoint()->shutdown(Seconds{10});
+            }
+        }
     });
 
     setupSignalHandlers();
@@ -414,11 +418,18 @@ int bridgeMain(int argc, char** argv, char** envp) {
     startSignalProcessingThread(LogFileStatus::kNoLogFileToRotate);
 
     auto serviceContext = getGlobalServiceContext();
+    serviceContext->setServiceEntryPoint(std::make_unique<ServiceEntryPointBridge>(serviceContext));
+    serviceContext->setServiceExecutor(
+        std::make_unique<transport::ServiceExecutorSynchronous>(serviceContext));
+
+    fassert(50766, serviceContext->getServiceExecutor()->start());
+
     transport::TransportLayerASIO::Options opts;
-    opts.mode = mongo::transport::TransportLayerASIO::Options::kEgress;
+    opts.ipList = "0.0.0.0";
+    opts.port = mongoBridgeGlobalParams.port;
 
-    serviceContext->setTransportLayer(
-        std::make_unique<mongo::transport::TransportLayerASIO>(opts, nullptr));
+    serviceContext->setTransportLayer(std::make_unique<mongo::transport::TransportLayerASIO>(
+        opts, serviceContext->getServiceEntryPoint()));
     auto tl = serviceContext->getTransportLayer();
     if (!tl->setup().isOK()) {
         log() << "Error setting up transport layer";
@@ -431,12 +442,7 @@ int bridgeMain(int argc, char** argv, char** envp) {
     }
 
     serviceContext->notifyStartupComplete();
-
-    listener = stdx::make_unique<BridgeListener>();
-    listener->setupSockets();
-    listener->initAndListen();
-
-    return EXIT_CLEAN;
+    return waitForShutdown();
 }
 
 }  // namespace mongo