SERVER-24162 Integrate TransportLayer

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

1 files changed, 20 insertions, 11 deletions

diff --git a/src/mongo/db/dbmessage.cpp b/src/mongo/db/dbmessage.cpp
index 67a5548dbfb..f16ec2b6276 100644
--- a/src/mongo/db/dbmessage.cpp
+++ b/src/mongo/db/dbmessage.cpp
@@ -30,7 +30,9 @@
 #include "mongo/platform/basic.h"
 
 #include "mongo/db/dbmessage.h"
+#include "mongo/db/operation_context.h"
 #include "mongo/platform/strnlen.h"
+#include "mongo/transport/session.h"
 
 namespace mongo {
 
@@ -173,20 +175,23 @@ OpQueryReplyBuilder::OpQueryReplyBuilder() : _buffer(32768) {
     _buffer.skip(sizeof(QueryResult::Value));
 }
 
-void OpQueryReplyBuilder::send(AbstractMessagingPort* destination,
+void OpQueryReplyBuilder::send(transport::Session* session,
                                int queryResultFlags,
-                               Message& requestMsg,
+                               const Message& requestMsg,
                                int nReturned,
                                int startingFrom,
                                long long cursorId) {
     Message response;
     putInMessage(&response, queryResultFlags, nReturned, startingFrom, cursorId);
-    destination->reply(requestMsg, response, requestMsg.header().getId());
+
+    response.header().setId(nextMessageId());
+    response.header().setResponseToMsgId(requestMsg.header().getId());
+
+    uassertStatusOK(session->sinkMessage(response).wait());
 }
 
-void OpQueryReplyBuilder::sendCommandReply(AbstractMessagingPort* destination,
-                                           Message& requestMsg) {
-    send(destination, /*queryFlags*/ 0, requestMsg, /*nReturned*/ 1);
+void OpQueryReplyBuilder::sendCommandReply(transport::Session* session, const Message& requestMsg) {
+    send(session, /*queryFlags*/ 0, requestMsg, /*nReturned*/ 1);
 }
 
 void OpQueryReplyBuilder::putInMessage(
@@ -202,7 +207,7 @@ void OpQueryReplyBuilder::putInMessage(
 }
 
 void replyToQuery(int queryResultFlags,
-                  AbstractMessagingPort* p,
+                  transport::Session* session,
                   Message& requestMsg,
                   const void* data,
                   int size,
@@ -211,15 +216,19 @@ void replyToQuery(int queryResultFlags,
                   long long cursorId) {
     OpQueryReplyBuilder reply;
     reply.bufBuilderForResults().appendBuf(data, size);
-    reply.send(p, queryResultFlags, requestMsg, nReturned, startingFrom, cursorId);
+    reply.send(session, queryResultFlags, requestMsg, nReturned, startingFrom, cursorId);
 }
 
 void replyToQuery(int queryResultFlags,
-                  AbstractMessagingPort* p,
+                  transport::Session* session,
                   Message& requestMsg,
                   const BSONObj& responseObj) {
-    replyToQuery(
-        queryResultFlags, p, requestMsg, (void*)responseObj.objdata(), responseObj.objsize(), 1);
+    replyToQuery(queryResultFlags,
+                 session,
+                 requestMsg,
+                 (void*)responseObj.objdata(),
+                 responseObj.objsize(),
+                 1);
 }
 
 void replyToQuery(int queryResultFlags, Message& m, DbResponse& dbresponse, BSONObj obj) {