[pulsar-client-cpp] fix snappy compressor compile error (#4972)

Fixes #4720 
### Motivation

for now, the snappy compressor can not be compiled successfully,
which raise undeclared identifier and no matching function errors.

### Modifications

fix compile errors, use snappy c++ apis.

pulsar-client-cpp/docker-build.sh

@@ -43,4 +43,4 @@ if [ "$1" != "skip-clean" ]; then
 	find . -name CMakeFiles | xargs rm -rf
 fi
 
-$DOCKER_CMD bash -c "cd /pulsar/pulsar-client-cpp && cmake . $CMAKE_ARGS && make check-format && make -j8"
+$DOCKER_CMD bash -c "cd /pulsar/pulsar-client-cpp && ./docker-lib-check.sh && cmake . $CMAKE_ARGS && make check-format && make -j8"

pulsar-client-cpp/docker-lib-check.sh

+#!/bin/bash
+#
+# 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.
+#
+
+if ! dpkg -l libsnappy-dev; then
+    apt-get install -y libsnappy-dev
+fi

pulsar-client-cpp/docker-tests.sh

@@ -58,4 +58,4 @@ done
 
 # Start 2 Pulsar standalone instances (one with TLS and one without)
 # and execute the tests
-$DOCKER_CMD bash -c "cd /pulsar/pulsar-client-cpp && ./run-unit-tests.sh ${tests}"
+$DOCKER_CMD bash -c "cd /pulsar/pulsar-client-cpp && ./docker-lib-check.sh && ./run-unit-tests.sh ${tests}"

pulsar-client-cpp/lib/CompressionCodecSnappy.cc

@@ -20,39 +20,28 @@
 
 #if HAS_SNAPPY
 #include <snappy.h>
-#include "snappy-c.h"
+#include <snappy-sinksource.h>
 
 namespace pulsar {
 
 SharedBuffer CompressionCodecSnappy::encode(const SharedBuffer& raw) {
     // Get the max size of the compressed data and allocate a buffer to hold it
-    int maxCompressedSize = snappy_max_compressed_length(raw.readableBytes());
-    SharedBuffer compressed = SharedBuffer::allocate(maxCompressedSize);
-
-    unsigned long bytesWritten = maxCompressedSize;
-
-    snappy_status status =
-        snappy_compress(raw.data(), raw.readableBytes(), compressed.mutableData(), &bytesWritten);
-
-    if (status != SNAPPY_OK) {
-        LOG_ERROR("Failed to compress to snappy. res=" << res);
-        abort();
-    }
-
-    compressed.bytesWritten(bytesWritten);
-
+    size_t maxCompressedLength = snappy::MaxCompressedLength(raw.readableBytes());
+    SharedBuffer compressed = SharedBuffer::allocate(static_cast<const uint32_t>(maxCompressedLength));
+    snappy::ByteArraySource source(raw.data(), raw.readableBytes());
+    snappy::UncheckedByteArraySink sink(compressed.mutableData());
+    size_t compressedSize = snappy::Compress(&source, &sink);
+    compressed.setWriterIndex(static_cast<uint32_t>(compressedSize));
     return compressed;
 }
 
 bool CompressionCodecSnappy::decode(const SharedBuffer& encoded, uint32_t uncompressedSize,
                                     SharedBuffer& decoded) {
-    SharedBuffer decompressed = SharedBuffer::allocate(uncompressedSize);
-
-    snappy_status status = snappy_uncompress(encoded.data(), encoded.readableBytes(),
-                                             decompressed.mutableData(), uncompressedSize);
-
-    if (status == SNAPPY_OK) {
-        decoded = decompressed;
+    SharedBuffer uncompressed = SharedBuffer::allocate(uncompressedSize);
+    snappy::ByteArraySource source(encoded.data(), encoded.readableBytes());
+    snappy::UncheckedByteArraySink sink(uncompressed.mutableData());
+    if (snappy::Uncompress(&source, &sink)) {
+        decoded = uncompressed;
         decoded.setWriterIndex(uncompressedSize);
         return true;
     } else {

pulsar-client-cpp/lib/CompressionCodecSnappy.h

@@ -22,7 +22,7 @@
 
 namespace pulsar {
 
-class CompressionCodecSnappy : public CompressionCodec {
+class PULSAR_PUBLIC CompressionCodecSnappy : public CompressionCodec {
    public:
     SharedBuffer encode(const SharedBuffer& raw);
 

pulsar-client-cpp/tests/CompressionCodecSnappyTest.cc

+/**
+ * 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/CompressionCodecSnappy.h"
+
+using namespace pulsar;
+
+TEST(CompressionCodecSnappyTest, testEncodeAndDecode) {
+    CompressionCodecSnappy compressionCodecSnappy;
+    char data[] = "snappy compression compresses snappy";
+    size_t sz = sizeof(data);
+    SharedBuffer source = SharedBuffer::wrap(data, sz);
+    SharedBuffer compressed = compressionCodecSnappy.encode(source);
+    ASSERT_GT(compressed.readableBytes(), 0);
+
+    SharedBuffer uncompressed;
+    bool res = compressionCodecSnappy.decode(compressed, static_cast<uint32_t>(sz), uncompressed);
+    ASSERT_TRUE(res);
+    ASSERT_EQ(uncompressed.readableBytes(), sz);
+    ASSERT_STREQ(data, uncompressed.data());
+}