[GitHub] merlimat closed pull request #2854: Fixed ZLib decompression in C++ client

[GitBox <gi...@apache.org>]

merlimat closed pull request #2854: Fixed ZLib decompression in C++ client
URL: https://github.com/apache/pulsar/pull/2854
 
 
   

This is a PR merged from a forked repository.
As GitHub hides the original diff on merge, it is displayed below for
the sake of provenance:

As this is a foreign pull request (from a fork), the diff is supplied
below (as it won't show otherwise due to GitHub magic):

diff --git a/pulsar-client-cpp/lib/CompressionCodec.h b/pulsar-client-cpp/lib/CompressionCodec.h
index 2fc6ead9e8..c67fce6bc5 100644
--- a/pulsar-client-cpp/lib/CompressionCodec.h
+++ b/pulsar-client-cpp/lib/CompressionCodec.h
@@ -28,6 +28,9 @@
 
 #include <map>
 
+// Make symbol visible to unit tests
+#pragma GCC visibility push(default)
+
 using namespace pulsar;
 namespace pulsar {
 
@@ -86,4 +89,6 @@ class CompressionCodecNone : public CompressionCodec {
 };
 }  // namespace pulsar
 
+#pragma GCC visibility pop
+
 #endif /* LIB_COMPRESSIONCODEC_H_ */
diff --git a/pulsar-client-cpp/lib/CompressionCodecZLib.cc b/pulsar-client-cpp/lib/CompressionCodecZLib.cc
index 0ddc644ceb..657c5488e9 100644
--- a/pulsar-client-cpp/lib/CompressionCodecZLib.cc
+++ b/pulsar-client-cpp/lib/CompressionCodecZLib.cc
@@ -28,13 +28,13 @@ DECLARE_LOG_OBJECT()
 
 namespace pulsar {
 
-SharedBuffer CompressionCodecZLib::encode(const SharedBuffer& raw) {
+SharedBuffer CompressionCodecZLib::encode(const SharedBuffer &raw) {
     // Get the max size of the compressed data and allocate a buffer to hold it
     int maxCompressedSize = compressBound(raw.readableBytes());
     SharedBuffer compressed = SharedBuffer::allocate(maxCompressedSize);
 
     unsigned long bytesWritten = maxCompressedSize;
-    int res = compress((Bytef*)compressed.mutableData(), &bytesWritten, (const Bytef*)raw.data(),
+    int res = compress((Bytef *)compressed.mutableData(), &bytesWritten, (const Bytef *)raw.data(),
                        raw.readableBytes());
     if (res != Z_OK) {
         LOG_ERROR("Failed to compress buffer. res=" << res);
@@ -45,17 +45,44 @@ SharedBuffer CompressionCodecZLib::encode(const SharedBuffer& raw) {
     return compressed;
 }
 
-bool CompressionCodecZLib::decode(const SharedBuffer& encoded, uint32_t uncompressedSize,
-                                  SharedBuffer& decoded) {
-    SharedBuffer decompressed = SharedBuffer::allocate(uncompressedSize);
+static bool buffer_uncompress(const char *compressedBuffer, unsigned long compressedSize, char *resultBuffer,
+                              uint32_t uncompressedSize) {
+    z_stream stream;
+    stream.next_in = (Bytef *)compressedBuffer;
+    stream.avail_in = compressedSize;
+    stream.zalloc = NULL;
+    stream.zfree = NULL;
+    stream.opaque = NULL;
+
+    int res = inflateInit2(&stream, MAX_WBITS);
+    if (res != Z_OK) {
+        LOG_ERROR("Failed to initialize inflate stream: " << res);
+        return false;
+    }
+
+    stream.next_out = (Bytef *)resultBuffer;
+    stream.avail_out = uncompressedSize;
+
+    res = inflate(&stream, Z_PARTIAL_FLUSH);
+    inflateEnd(&stream);
 
-    unsigned long bytesUncompressed = uncompressedSize;
-    int res = uncompress((Bytef*)decompressed.mutableData(), &bytesUncompressed, (Bytef*)encoded.data(),
-                         encoded.readableBytes());
+    if (res == Z_OK || res == Z_STREAM_END) {
+        return true;
+    } else {
+        LOG_ERROR("Failed to decompress zlib buffer: " << res << " -- compressed size: " << compressedSize
+                                                       << " -- uncompressed size: " << uncompressedSize);
+        return false;
+    }
+}
+
+bool CompressionCodecZLib::decode(const SharedBuffer &encoded, uint32_t uncompressedSize,
+                                  SharedBuffer &decoded) {
+    SharedBuffer decompressed = SharedBuffer::allocate(uncompressedSize);
 
-    decompressed.bytesWritten(bytesUncompressed);
-    if (res == Z_OK) {
+    if (buffer_uncompress(encoded.data(), encoded.readableBytes(), decompressed.mutableData(),
+                          uncompressedSize)) {
         decoded = decompressed;
+        decoded.setWriterIndex(uncompressedSize);
         return true;
     } else {
         return false;
diff --git a/pulsar-client-cpp/lib/CompressionCodecZLib.h b/pulsar-client-cpp/lib/CompressionCodecZLib.h
index 52e43ff05f..08fd0c92c9 100644
--- a/pulsar-client-cpp/lib/CompressionCodecZLib.h
+++ b/pulsar-client-cpp/lib/CompressionCodecZLib.h
@@ -23,6 +23,9 @@
 #include <zlib.h>
 #include <boost/thread/mutex.hpp>
 
+// Make symbol visible to unit tests
+#pragma GCC visibility push(default)
+
 namespace pulsar {
 
 class CompressionCodecZLib : public CompressionCodec {
@@ -31,6 +34,9 @@ class CompressionCodecZLib : public CompressionCodec {
 
     bool decode(const SharedBuffer& encoded, uint32_t uncompressedSize, SharedBuffer& decoded);
 };
+
 }  // namespace pulsar
 
+#pragma GCC visibility pop
+
 #endif /* LIB_COMPRESSIONCODECZLIB_H_ */
diff --git a/pulsar-client-cpp/perf/PerfProducer.cc b/pulsar-client-cpp/perf/PerfProducer.cc
index 596afd5ccb..77788f277d 100644
--- a/pulsar-client-cpp/perf/PerfProducer.cc
+++ b/pulsar-client-cpp/perf/PerfProducer.cc
@@ -65,6 +65,7 @@ struct Arguments {
     long batchingMaxPublishDelayMs;
     std::string encKeyName;
     std::string encKeyValueFile;
+    std::string compression;
 };
 
 namespace pulsar {
@@ -259,6 +260,9 @@ int main(int argc, char** argv) {
     ("batch-size", po::value<unsigned int>(&args.batchingMaxMessages)->default_value(1),
             "If batch size == 1 then batching is disabled. Default batch size == 1") //
 
+    ("compression", po::value<std::string>(&args.compression)->default_value(""),
+             "Compression can be either 'zlib' or 'lz4'. Default is no compression") //
+
     ("max-batch-size-in-bytes", po::value<long>(&args.batchingMaxAllowedSizeInBytes)->default_value(128 * 1024),
             "Use only is batch-size > 1, Default is 128 KB") //
 
@@ -329,6 +333,15 @@ int main(int argc, char** argv) {
         producerConf.setBatchingMaxPublishDelayMs(args.batchingMaxPublishDelayMs);
     }
 
+    if (args.compression == "zlib") {
+        producerConf.setCompressionType(CompressionZLib);
+    } else if (args.compression == "lz4") {
+        producerConf.setCompressionType(CompressionLZ4);
+    } else if (!args.compression.empty()) {
+        LOG_WARN("Invalid compression type: " << args.compression);
+        return -1;
+    }
+
     // Block if queue is full else we will start seeing errors in sendAsync
     producerConf.setBlockIfQueueFull(true);
     boost::shared_ptr<EncKeyReader> keyReader = boost::make_shared<EncKeyReader>(args.encKeyValueFile);
diff --git a/pulsar-client-cpp/tests/ZLibCompressionTest.cc b/pulsar-client-cpp/tests/ZLibCompressionTest.cc
new file mode 100644
index 0000000000..c510db159f
--- /dev/null
+++ b/pulsar-client-cpp/tests/ZLibCompressionTest.cc
@@ -0,0 +1,65 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you 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 <gtest/gtest.h>
+#include <lib/CompressionCodecZLib.h>
+
+using namespace pulsar;
+
+TEST(ZLibCompressionTest, compressDecompress) {
+    CompressionCodecZLib codec;
+
+    std::string payload = "Payload to compress";
+    SharedBuffer compressed = codec.encode(SharedBuffer::copy(payload.c_str(), payload.size()));
+
+    SharedBuffer uncompressed;
+    bool res = codec.decode(compressed, payload.size(), uncompressed);
+    ASSERT_TRUE(res);
+    ASSERT_EQ(payload, std::string(uncompressed.data(), uncompressed.readableBytes()));
+}
+
+// Java and C++ are using different ZLib settings when compressing, so the resulting
+// compressed blobs are slightly different. Both should lead to the same result when
+// decompressing
+TEST(ZLibCompressionTest, decodeCppCompressed) {
+    CompressionCodecZLib codec;
+
+    const uint8_t compressed[] = {0x78, 0x9c, 0x63, 0x60, 0x80, 0x01, 0x00, 0x00, 0x0a, 0x00, 0x01};
+
+    SharedBuffer uncompressed;
+    uint32_t uncompressedSize = 10;
+
+    bool res = codec.decode(SharedBuffer::copy((const char*)compressed, sizeof(compressed)), uncompressedSize,
+                            uncompressed);
+    ASSERT_TRUE(res);
+    ASSERT_EQ(uncompressedSize, uncompressed.readableBytes());
+}
+
+TEST(ZLibCompressionTest, decodeJavaCompressed) {
+    CompressionCodecZLib codec;
+
+    const uint8_t compressed[] = {0x78, 0x9c, 0x62, 0x60, 0x80, 0x01, 0x00, 0x00, 0x00, 0x00, 0xff, 0xff};
+
+    SharedBuffer uncompressed;
+    uint32_t uncompressedSize = 10;
+
+    bool res = codec.decode(SharedBuffer::copy((const char*)compressed, sizeof(compressed)), uncompressedSize,
+                            uncompressed);
+    ASSERT_TRUE(res);
+    ASSERT_EQ(uncompressedSize, uncompressed.readableBytes());
+}


 

----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
 
For queries about this service, please contact Infrastructure at:
users@infra.apache.org


With regards,
Apache Git Services