Moved databuffer and compresssion to vespalib

2 files changed, 15 insertions, 2 deletions

diff --git a/storage/src/vespa/storage/config/stor-communicationmanager.def b/storage/src/vespa/storage/config/stor-communicationmanager.def
index 23c713d41f9..2c00dbe2f37 100644
--- a/storage/src/vespa/storage/config/stor-communicationmanager.def
+++ b/storage/src/vespa/storage/config/stor-communicationmanager.def
@@ -17,3 +17,11 @@ mbus_content_node_max_pending_count int default=0
 mbus_distributor_node_max_pending_size int default=0
 mbus_content_node_max_pending_size int default=0
 
+# Minimum size of packets to compress (0 means no compression)
+mbus.compress.limit int default=1024
+
+## Compression level for packets
+mbus.compress.level int default=3
+
+## Compression type for packets.
+mbus.compress.type enum {NONE, LZ4, ZSTD} default=LZ4
diff --git a/storage/src/vespa/storage/storageserver/communicationmanager.cpp b/storage/src/vespa/storage/storageserver/communicationmanager.cpp
index aac52550fa0..c90b18038c2 100644
--- a/storage/src/vespa/storage/storageserver/communicationmanager.cpp
+++ b/storage/src/vespa/storage/storageserver/communicationmanager.cpp
@@ -8,6 +8,7 @@
 #include <vespa/documentapi/messagebus/messages/wrongdistributionreply.h>
 #include <vespa/storageapi/message/state.h>
 #include <vespa/messagebus/rpcmessagebus.h>
+#include <vespa/messagebus/network/rpcnetworkparams.h>
 #include <vespa/messagebus/emptyreply.h>
 #include <vespa/vespalib/stllike/asciistream.h>
 #include <vespa/vespalib/util/stringfmt.h>
@@ -362,8 +363,7 @@ void CommunicationManager::onClose()
 }
 
 void
-CommunicationManager::configureMessageBusLimits(
-        const CommunicationManagerConfig& cfg)
+CommunicationManager::configureMessageBusLimits(const CommunicationManagerConfig& cfg)
 {
     const bool isDist(_component.getNodeType() == lib::NodeType::DISTRIBUTOR);
     auto& mbus(_mbus->getMessageBus());
@@ -404,6 +404,11 @@ void CommunicationManager::configure(std::unique_ptr<CommunicationManagerConfig>
             params.setListenPort(config->mbusport);
         }
 
+        using CompressionConfig = vespalib::compression::CompressionConfig;
+        CompressionConfig::Type compressionType = CompressionConfig::toType(
+                CommunicationManagerConfig::Mbus::Compress::getTypeName(config->mbus.compress.type).c_str());
+        params.setCompressionConfig(CompressionConfig(compressionType, config->mbus.compress.level,
+                                                      90, config->mbus.compress.limit));
         // Configure messagebus here as we for legacy reasons have
         // config here.
         _mbus = std::make_unique<mbus::RPCMessageBus>(