SERVER-25267 SERVER-25265 Integrate compression with networking code

and snappy compressor

2 files changed, 32 insertions, 0 deletions

diff --git a/src/mongo/tools/bridge.cpp b/src/mongo/tools/bridge.cpp
index d37751fe537..6299e26b81c 100644
--- a/src/mongo/tools/bridge.cpp
+++ b/src/mongo/tools/bridge.cpp
@@ -128,6 +128,7 @@ public:
 
         Message request;
         Message response;
+        MessageCompressorManager compressorManager;
 
         while (true) {
             try {
@@ -138,6 +139,16 @@ public:
                     break;
                 }
 
+                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());
+                }
+
                 std::unique_ptr<rpc::RequestInterface> cmdRequest;
                 if (request.operation() == dbQuery || request.operation() == dbCommand) {
                     cmdRequest = rpc::makeRequest(&request);
@@ -245,6 +256,16 @@ public:
                         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()) {
diff --git a/src/mongo/tools/mongobridge_options_init.cpp b/src/mongo/tools/mongobridge_options_init.cpp
index 5207fc0b49d..dadd88b24b4 100644
--- a/src/mongo/tools/mongobridge_options_init.cpp
+++ b/src/mongo/tools/mongobridge_options_init.cpp
@@ -30,6 +30,7 @@
 
 #include <iostream>
 
+#include "mongo/transport/message_compressor_registry.h"
 #include "mongo/util/exit_code.h"
 #include "mongo/util/options_parser/startup_option_init.h"
 #include "mongo/util/options_parser/startup_options.h"
@@ -37,6 +38,9 @@
 
 namespace mongo {
 MONGO_GENERAL_STARTUP_OPTIONS_REGISTER(MongoBridgeOptions)(InitializerContext* context) {
+    auto ret = addMessageCompressionOptions(&moe::startupOptions, false);
+    if (!ret.isOK())
+        return ret;
     return addMongoBridgeOptions(&moe::startupOptions);
 }
 
@@ -58,6 +62,13 @@ MONGO_STARTUP_OPTIONS_STORE(MongoBridgeOptions)(InitializerContext* context) {
         std::cerr << "try '" << context->args()[0] << " --help' for more information" << std::endl;
         quickExit(EXIT_BADOPTIONS);
     }
+
+    ret = storeMessageCompressionOptions(moe::startupOptionsParsed);
+    if (!ret.isOK()) {
+        std::cerr << ret.toString() << std::endl;
+        quickExit(EXIT_BADOPTIONS);
+    }
+
     return Status::OK();
 }
 }