Add support for message compression

CMakeLists.txt

@@ -1,6 +1,8 @@
-cmake_minimum_required(VERSION 3.25) 
+cmake_minimum_required(VERSION 3.25)
 project(rmq LANGUAGES C CXX)
 
+option(ENABLE_COMPRESSION "Enable zstd compression support" ON)
+
 set(CMAKE_EXPORT_COMPILE_COMMANDS 1)
 enable_testing()
 

README.md

@@ -31,7 +31,7 @@ This library has been built from experience learned while supporting other Rabbi
     + `rmqcpp` always declares topology when creating consumers & producers, as per RabbitMQ best practices.
 4. ✉ Reliable Message Delivery 'on' by default
     + Publisher confirmations. This ensures clients are aware when messages are owned by RabbitMQ, and avoids messages being silently black holed.
-    + Consumer acknowledgements. Switching these on manually helps avoid messages being silently dropped during restart/outages, as would be the case with 'autoack'. 
+    + Consumer acknowledgements. Switching these on manually helps avoid messages being silently dropped during restart/outages, as would be the case with 'autoack'.
     + Durable queues and persistent delivery mode ensure messages always persist during broker restarts and total datacenter shutdowns.
     + [Mandatory flag](https://www.rabbitmq.com/amqp-0-9-1-reference.html#basic.publish.mandatory) is defaulted to 'true' for all messages to ensure none are silently dropped due to missed bindings.
     + All of the above properties are used by default. Publisher confirms and consumer acknowledgements are required.
@@ -58,7 +58,7 @@ graph TD;
     rmqp-->rmqt;
 ```
 
-**Library** | **Purpose** | **Examples** 
+**Library** | **Purpose** | **Examples**
  :--- | :---- | :----
 [rmqp](src/rmq/rmqp) | RabbitMQ library interface (protocol) | Interfaces used to allow testing/mocking of `rmqcpp` applications
 [rmqa](src/rmq/rmqa) | RabbitMQ Library interface implementation | The main concrete objects used by applications
@@ -70,7 +70,7 @@ graph TD;
 
 ## Quick Start
 The quickest way to get started is to take a look at our integration tests and sample 'hello world' program, which is possible by following the Docker [Build](#building) steps and then: from the interactive shell window running `./build/examples/helloworld/rmqhelloworld_producer`
- 
+
 
 ## Usage
 
@@ -195,7 +195,7 @@ consumer->cancelAndDrain();
 
 ## Documentation
 Doxygen generated API documentation can be found [here](https://bloomberg.github.io/rmqcpp/index.html)
- 
+
 ## Building
 
 ### Prerequisites
@@ -206,6 +206,8 @@ Doxygen generated API documentation can be found [here](https://bloomberg.github
 
 There are build configuration options which can be specified using the environment variable `CMAKE_PRESET` (choose from configurations in `CMakePresets.json`) - eg. `export CMAKE_PRESET=macos-arm64-vcpkg`.
 
+`zstd` compression support is enabled by default, and requires the `zstd` library to be installed. To disable this, pass the `-DENABLE_COMPRESSION=OFF` option to cmake.
+
 ### Build Steps
 Once the prerequisites are configured:
 
@@ -221,8 +223,8 @@ We also provide Dockerfiles for building and running this in an isolated
 environment. If you don't wish to get vcpkg set up on your build machine, this can be an alternative
 quick way to get started.
 
-1. `make docker-setup` - Build required base images and setup vcpkg, prerequisite for running commands below 
-2. `make docker-build` - Build rmqcpp in the container using vcpkg 
+1. `make docker-setup` - Build required base images and setup vcpkg, prerequisite for running commands below
+2. `make docker-build` - Build rmqcpp in the container using vcpkg
 3. `make docker-unit`  - Build rmqcpp and run unit tests in the container
 4. `make docker-shell` - Get an interactive shell within the build environment container
 

dockerfiles/dev.Dockerfile

@@ -5,11 +5,12 @@ RUN apt-get update && apt-get install -y \
     clang-format \
     cmake \
     curl \
-    gcc \ 
+    gcc \
     gdb \
     git \
     libboost-dev \
     libssl-dev \
+    libzstd-dev \
     net-tools \
     netcat-traditional \
     ninja-build \
@@ -24,7 +25,7 @@ RUN apt-get update && apt-get install -y \
     zip \
     && rm -rf /var/lib/apt/lists/*
 
-ENV VCPKG_FORCE_SYSTEM_BINARIES=1 
+ENV VCPKG_FORCE_SYSTEM_BINARIES=1
 
 # clone and install vcpkg
 RUN git clone https://github.com/Microsoft/vcpkg.git /build/vcpkg && \

licenses-binary/LICENSE-zstd.txt

@@ -0,0 +1,30 @@
+BSD License
+
+For Zstandard software
+
+Copyright (c) Meta Platforms, Inc. and affiliates. All rights reserved.
+
+Redistribution and use in source and binary forms, with or without modification,
+are permitted provided that the following conditions are met:
+
+ * Redistributions of source code must retain the above copyright notice, this
+   list of conditions and the following disclaimer.
+
+ * Redistributions in binary form must reproduce the above copyright notice,
+   this list of conditions and the following disclaimer in the documentation
+   and/or other materials provided with the distribution.
+
+ * Neither the name Facebook, nor Meta, nor the names of its contributors may
+   be used to endorse or promote products derived from this software without
+   specific prior written permission.
+
+THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND
+ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
+WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
+DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE FOR
+ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES
+(INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES;
+LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON
+ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+(INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS
+SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.

licenses/LICENSE-zstd.txt

@@ -0,0 +1,30 @@
+BSD License
+
+For Zstandard software
+
+Copyright (c) Meta Platforms, Inc. and affiliates. All rights reserved.
+
+Redistribution and use in source and binary forms, with or without modification,
+are permitted provided that the following conditions are met:
+
+ * Redistributions of source code must retain the above copyright notice, this
+   list of conditions and the following disclaimer.
+
+ * Redistributions in binary form must reproduce the above copyright notice,
+   this list of conditions and the following disclaimer in the documentation
+   and/or other materials provided with the distribution.
+
+ * Neither the name Facebook, nor Meta, nor the names of its contributors may
+   be used to endorse or promote products derived from this software without
+   specific prior written permission.
+
+THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND
+ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
+WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
+DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE FOR
+ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES
+(INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES;
+LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON
+ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+(INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS
+SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.

src/CMakeLists.txt

@@ -1,5 +1,8 @@
 find_package(Threads REQUIRED) # CMake 3.26-rc3 Bug https://gitlab.kitware.com/cmake/cmake/-/issues/24505
 find_package(ZLIB REQUIRED)
+if (ENABLE_COMPRESSION)
+    find_package(zstd CONFIG REQUIRED)
+endif()
 find_package(Boost REQUIRED)
 set(OPENSSL_USE_STATIC_LIBS TRUE)
 find_package(OpenSSL REQUIRED)

src/rmq/CMakeLists.txt

@@ -15,6 +15,9 @@ add_library(rmq STATIC
 )
 
 target_link_libraries(rmq PUBLIC bsl bdl bal ZLIB::ZLIB OpenSSL::Crypto OpenSSL::SSL)
+if (ENABLE_COMPRESSION)
+    target_link_libraries(rmq PUBLIC zstd::libzstd_static)
+endif()
 
 get_target_property(OPENSSL_TARGET_TYPE OpenSSL::SSL TYPE)
 if(OPENSSL_CRYPTO_LIBRARY MATCHES "\\.a$")
@@ -42,19 +45,19 @@ install(
         COMPONENT librmq-dev
 )
 
-install(EXPORT rmqcppTargets 
-        FILE rmqcppTargets.cmake 
-        DESTINATION share/rmqcpp 
+install(EXPORT rmqcppTargets
+        FILE rmqcppTargets.cmake
+        DESTINATION share/rmqcpp
         NAMESPACE rmqcpp::)
 
 include(CMakePackageConfigHelpers)
-configure_package_config_file(${CMAKE_CURRENT_SOURCE_DIR}/Config.cmake.in 
+configure_package_config_file(${CMAKE_CURRENT_SOURCE_DIR}/Config.cmake.in
                               "${CMAKE_CURRENT_BINARY_DIR}/rmqcppConfig.cmake"
-                              INSTALL_DESTINATION share/rmqcpp) 
+                              INSTALL_DESTINATION share/rmqcpp)
 install(FILES "${CMAKE_CURRENT_BINARY_DIR}/rmqcppConfig.cmake" DESTINATION share/rmqcpp)
 
 # Emit some metadata required internally
-set(RMQ_PC_DEP_NAMES bsl bdl bal openssl)
+set(RMQ_PC_DEP_NAMES bsl bdl bal openssl zstd)
 find_package(GenBDEMetadata QUIET)
 if (GenBDEMetadata_FOUND)
     gen_bde_metadata(PACKAGE_GROUP rmq INSTALL_COMPONENT librmq-dev DEPS "${RMQ_PC_DEP_NAMES}")

src/rmq/rmqa/CMakeLists.txt

@@ -1,4 +1,6 @@
-add_library(rmqa OBJECT 
+add_library(rmqa OBJECT
+    rmqa_compressiontransformer.cpp
+    rmqa_compressiontransformerimpl.cpp
     rmqa_consumer.cpp
     rmqa_consumerimpl.cpp
     rmqa_connectionimpl.cpp
@@ -25,12 +27,16 @@ target_link_libraries(rmqa PUBLIC
     bsl
     bdl
     bal
-    rmqt 
+    rmqt
     rmqp
     rmqamqp
     rmqio
     rmqamqpt
 )
+if (ENABLE_COMPRESSION)
+    target_link_libraries(rmqa PUBLIC zstd::libzstd_static)
+    target_compile_definitions(rmqa PRIVATE RMQCPP_ENABLE_COMPRESSION)
+endif()
 if( "${CMAKE_CXX_COMPILER_ID}" STREQUAL "SunPro" )
     # _RWSTD_ALLOCATOR tells the solaris <memory> header to define a std::allocator
     # which conforms better to the C++ standard, which is expected by Boost. Without

src/rmq/rmqa/rmqa_compressiontransformer.cpp

@@ -0,0 +1,58 @@
+// Copyright 2025 Bloomberg Finance L.P.
+// SPDX-License-Identifier: Apache-2.0
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+#include <bsl_memory.h>
+#include <bsl_vector.h>
+#include <bslma_managedptr.h>
+#include <rmqa_compressiontransformer.h>
+#include <rmqa_compressiontransformerimpl.h>
+#include <rmqp_messagetransformer.h>
+#include <rmqt_properties.h>
+
+namespace BloombergLP {
+namespace rmqa {
+
+rmqt::Result<rmqp::MessageTransformer> CompressionTransformer::create()
+{
+    return rmqt::Result<rmqp::MessageTransformer>(
+        bsl::make_shared<CompressionTransformerImpl>());
+}
+
+CompressionTransformer::CompressionTransformer(
+    bslma::ManagedPtr<rmqp::MessageTransformer>& impl)
+: d_impl(impl)
+{
+}
+
+CompressionTransformer::~CompressionTransformer() {}
+
+rmqt::Result<bool>
+CompressionTransformer::transform(bsl::shared_ptr<bsl::vector<uint8_t> >& data,
+                                  rmqt::Properties& props)
+{
+    return d_impl->transform(data, props);
+}
+
+rmqt::Result<> CompressionTransformer::inverseTransform(
+    bsl::shared_ptr<bsl::vector<uint8_t> >& data,
+    rmqt::Properties& props)
+{
+    return d_impl->inverseTransform(data, props);
+}
+
+bsl::string CompressionTransformer::name() const { return d_impl->name(); }
+
+} // namespace rmqa
+} // namespace BloombergLP

src/rmq/rmqa/rmqa_compressiontransformer.h

@@ -0,0 +1,98 @@
+// Copyright 2025 Bloomberg Finance L.P.
+// SPDX-License-Identifier: Apache-2.0
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+#ifndef INCLUDED_RMQA_COMPRESSIONTRANSFORMER
+#define INCLUDED_RMQA_COMPRESSIONTRANSFORMER
+
+#include <rmqp_messagetransformer.h>
+#include <rmqt_properties.h>
+#include <rmqt_result.h>
+
+#include <bsl_memory.h>
+#include <bsl_string.h>
+#include <bsl_vector.h>
+#include <bslma_managedptr.h>
+#include <bsls_keyword.h>
+
+namespace BloombergLP {
+namespace rmqa {
+
+/// \class CompressionTransformer
+/// \brief A message transformer that compresses messages using zstd
+///
+/// CompressionTransformers are typically bound to Producers and
+/// Consumers, to process messages before sending and after receiving them.
+class CompressionTransformer {
+  public:
+    // CREATORS
+    /// Create an instance of a CompressionTransformer that can be used
+    /// by a single thread to manage message compression.
+    /// The function may fail if insufficient memory is available to allocate
+    /// a compression context.
+    static rmqt::Result<rmqp::MessageTransformer> create();
+
+    explicit CompressionTransformer(
+        bslma::ManagedPtr<rmqp::MessageTransformer>& impl);
+
+    ~CompressionTransformer();
+
+    // MANIPULATORS
+    /// Transform the given `data` and `props` in-place into compressed form,
+    /// if possible. The resulting data will never be larger than the original
+    /// data, and the properties will be updated to indicate that the message
+    /// has been compressed.
+    ///
+    /// \param data  The data to be compressed
+    /// \param props The message properties
+    ///
+    /// \return true   Returned if the data was successfully compressed
+    /// \return false  Returned if the data is deemed to be not compressible.
+    ///                In this case, the data and properties will remain
+    ///                unchanged
+    /// \return error  Returned if the compression failed for any other
+    ///                reason. The data and properties will remain
+    ///                unchanged
+    rmqt::Result<bool> transform(bsl::shared_ptr<bsl::vector<uint8_t> >& data,
+                                 rmqt::Properties& props);
+
+    /// Decompresses the given `data` and `props` in-place, if they were
+    /// compressed. If the data is not compressed, it will remain unchanged.
+    ///
+    /// \param data  The data to be compressed
+    /// \param props The message properties
+    ///
+    /// \return true   Returned if the data was successfully decompressed
+    /// \return false  Returned if the decompression failed
+    rmqt::Result<>
+    inverseTransform(bsl::shared_ptr<bsl::vector<uint8_t> >& data,
+                     rmqt::Properties& props);
+
+    /// Returns the header name used by this transformer to indicate
+    /// that the message has been compressed.
+    bsl::string name() const;
+
+  private:
+    CompressionTransformer(const CompressionTransformer&) BSLS_KEYWORD_DELETED;
+    CompressionTransformer&
+    operator=(const CompressionTransformer&) BSLS_KEYWORD_DELETED;
+
+  private:
+    bslma::ManagedPtr<rmqp::MessageTransformer> d_impl;
+};
+
+} // namespace rmqa
+} // namespace BloombergLP
+
+#endif