Support bzip2 and gzip compress (#1399)

* support for #1398

Co-authored-by: 呈铭 <[email protected]>
Co-authored-by: evenliu <[email protected]>
Co-authored-by: coderabbitai[bot] <136622811+coderabbitai[bot]@users.noreply.github.com>

all/pom.xml

@@ -74,6 +74,7 @@
         <guava.version>32.0.0-jre</guava.version>
         <transmittable.version>2.12.1</transmittable.version>
         <fury.version>0.4.1</fury.version>
+        <commons_compress_version>1.25.0</commons_compress_version>
     </properties>
 
     <dependencies>
@@ -445,6 +446,11 @@
             <artifactId>fury-core</artifactId>
             <version>${fury.version}</version>
         </dependency>
+        <dependency>
+            <groupId>org.apache.commons</groupId>
+            <artifactId>commons-compress</artifactId>
+            <version>${commons_compress_version}</version>
+        </dependency>
 
     </dependencies>
 

bom/pom.xml

@@ -62,6 +62,7 @@
         <module.deploy.skip>true</module.deploy.skip>
         <!-- Fabric8 for Kubernetes -->
         <fabric8_kubernetes_version>6.9.2</fabric8_kubernetes_version>
+        <commons_compress_version>1.25.0</commons_compress_version>
     </properties>
 
     <dependencyManagement>
@@ -515,7 +516,6 @@
                 <artifactId>grpc-stub</artifactId>
                 <version>${grpc.version}</version>
             </dependency>
-
             <!-- Fabric8 for Kubernetes -->
             <dependency>
                 <groupId>io.fabric8</groupId>
@@ -529,6 +529,8 @@
                 <version>${fabric8_kubernetes_version}</version>
             </dependency>
 
+
+
             <!-- Test libs -->
             <dependency>
                 <groupId>org.apache.curator</groupId>
@@ -613,6 +615,12 @@
                 <version>0.16.0</version>
                 <scope>test</scope>
             </dependency>
+
+            <dependency>
+                <groupId>org.apache.commons</groupId>
+                <artifactId>commons-compress</artifactId>
+                <version>${commons_compress_version}</version>
+            </dependency>
         </dependencies>
     </dependencyManagement>
 

codec/codec-api/pom.xml

@@ -38,6 +38,11 @@
             <artifactId>junit</artifactId>
             <scope>test</scope>
         </dependency>
+
+        <dependency>
+            <groupId>org.apache.commons</groupId>
+            <artifactId>commons-compress</artifactId>
+        </dependency>
     </dependencies>
 
     <build>

codec/codec-api/src/main/java/com/alipay/sofa/rpc/codec/bzip2/Bzip2RpcCompressor.java

@@ -0,0 +1,76 @@
+/*
+ * 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.
+ */
+package com.alipay.sofa.rpc.codec.bzip2;
+
+import com.alipay.sofa.rpc.codec.Compressor;
+import com.alipay.sofa.rpc.ext.Extension;
+import org.apache.commons.compress.compressors.bzip2.BZip2CompressorInputStream;
+import org.apache.commons.compress.compressors.bzip2.BZip2CompressorOutputStream;
+
+import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
+
+/**
+ * bzip2 compressor, faster compression efficiency
+ *
+ * @author chengming
+ * @version Bzip2RpcCompressor.java, v 0.1 2024年02月28日 10:45 AM chengming
+ * @link https://commons.apache.org/proper/commons-compress/
+ */
+@Extension(value = "bzip2", code = 3)
+public class Bzip2RpcCompressor implements Compressor {
+
+    @Override
+    public byte[] compress(byte[] src) {
+        if (null == src || 0 == src.length) {
+            return new byte[0];
+        }
+
+        ByteArrayOutputStream out = new ByteArrayOutputStream();
+        BZip2CompressorOutputStream cos;
+        try {
+            cos = new BZip2CompressorOutputStream(out);
+            cos.write(src);
+            cos.close();
+        } catch (Exception e) {
+            throw new IllegalStateException(e);
+        }
+
+        return out.toByteArray();
+    }
+
+    @Override
+    public byte[] deCompress(byte[] src) {
+        if (null == src || 0 == src.length) {
+            return new byte[0];
+        }
+
+        ByteArrayOutputStream out = new ByteArrayOutputStream();
+        ByteArrayInputStream in = new ByteArrayInputStream(src);
+        try {
+            BZip2CompressorInputStream unZip = new BZip2CompressorInputStream(in);
+            byte[] buffer = new byte[2048];
+            int n;
+            while ((n = unZip.read(buffer)) >= 0) {
+                out.write(buffer, 0, n);
+            }
+        } catch (Exception e) {
+            throw new IllegalStateException(e);
+        }
+        return out.toByteArray();
+    }
+}

codec/codec-api/src/main/java/com/alipay/sofa/rpc/codec/gzip/GzipRpcCompressor.java

@@ -0,0 +1,70 @@
+/*
+ * 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.
+ */
+package com.alipay.sofa.rpc.codec.gzip;
+
+import com.alipay.sofa.rpc.codec.Compressor;
+import com.alipay.sofa.rpc.ext.Extension;
+
+import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
+import java.util.zip.GZIPInputStream;
+import java.util.zip.GZIPOutputStream;
+
+/**
+ * @author chengming
+ * @version GzipRpcCompressor.java, v 0.1 2024年02月28日 11:25 AM chengming
+ */
+@Extension(value = "gzip", code = 4)
+public class GzipRpcCompressor implements Compressor {
+
+    @Override
+    public byte[] compress(byte[] src) {
+        if (null == src || 0 == src.length) {
+            return new byte[0];
+        }
+
+        ByteArrayOutputStream byteOutStream = new ByteArrayOutputStream();
+        try (GZIPOutputStream gzipOutputStream = new GZIPOutputStream(byteOutStream)) {
+            gzipOutputStream.write(src);
+        } catch (Exception exception) {
+            throw new IllegalStateException(exception);
+        }
+
+        return byteOutStream.toByteArray();
+    }
+
+    @Override
+    public byte[] deCompress(byte[] src) {
+        if (null == src || 0 == src.length) {
+            return new byte[0];
+        }
+
+        ByteArrayInputStream byteInStream = new ByteArrayInputStream(src);
+        ByteArrayOutputStream byteOutStream = new ByteArrayOutputStream();
+        try (GZIPInputStream gzipInputStream = new GZIPInputStream(byteInStream)) {
+            int readByteNum;
+            byte[] bufferArr = new byte[256];
+            while ((readByteNum = gzipInputStream.read(bufferArr)) >= 0) {
+                byteOutStream.write(bufferArr, 0, readByteNum);
+            }
+        } catch (Exception exception) {
+            throw new IllegalStateException(exception);
+        }
+
+        return byteOutStream.toByteArray();
+    }
+}

codec/codec-api/src/main/resources/META-INF/services/sofa-rpc/com.alipay.sofa.rpc.codec.Compressor

@@ -1 +1,3 @@
-snappy=com.alipay.sofa.rpc.codec.snappy.SnappyRpcCompressor
+snappy=com.alipay.sofa.rpc.codec.snappy.SnappyRpcCompressor
+bzip2=com.alipay.sofa.rpc.codec.bzip2.Bzip2RpcCompressor
+gzip=com.alipay.sofa.rpc.codec.gzip.GzipRpcCompressor

codec/codec-api/src/test/java/com/alipay/sofa/rpc/codec/bzip2/Bzip2RpcCompressorTest.java

@@ -0,0 +1,53 @@
+/*
+ * 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.
+ */
+package com.alipay.sofa.rpc.codec.bzip2;
+
+import com.alipay.sofa.rpc.codec.Compressor;
+import com.alipay.sofa.rpc.ext.ExtensionLoaderFactory;
+import org.junit.Assert;
+import org.junit.Test;
+
+import java.io.UnsupportedEncodingException;
+
+/**
+ * @author chengming
+ * @version Bzip2RpcCompressorTest.java, v 0.1 2024年02月28日 2:19 PM chengming
+ */
+public class Bzip2RpcCompressorTest {
+
+    private static final String TEST_STR;
+
+    static {
+        StringBuilder builder = new StringBuilder();
+        int charNum = 1000000;
+        for (int i = 0; i < charNum; i++) {
+            builder.append("a");
+        }
+
+        TEST_STR = builder.toString();
+    }
+
+    @Test
+    public void testCompression() throws UnsupportedEncodingException {
+        Compressor compressor = ExtensionLoaderFactory.getExtensionLoader(Compressor.class).getExtension("bzip2");
+        Assert.assertTrue(compressor instanceof Bzip2RpcCompressor);
+
+        byte[] bs = compressor.compress(TEST_STR.getBytes("utf-8"));
+        String s1 = new String(compressor.deCompress(bs), "utf-8");
+        Assert.assertEquals(TEST_STR, s1);
+    }
+}

codec/codec-api/src/test/java/com/alipay/sofa/rpc/codec/gzip/GzipRpcCompressorTest.java

@@ -0,0 +1,54 @@
+/*
+ * 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.
+ */
+package com.alipay.sofa.rpc.codec.gzip;
+
+import com.alipay.sofa.rpc.codec.Compressor;
+import com.alipay.sofa.rpc.ext.ExtensionLoaderFactory;
+import org.junit.Assert;
+import org.junit.Test;
+
+import java.io.UnsupportedEncodingException;
+
+/**
+ * @author chengming
+ * @version GzipRpcCompressorTest.java, v 0.1 2024年02月28日 2:09 PM chengming
+ */
+public class GzipRpcCompressorTest {
+
+    private static final String TEST_STR;
+
+    static {
+        StringBuilder builder = new StringBuilder();
+        int charNum = 1000000;
+        for (int i = 0; i < charNum; i++) {
+            builder.append("a");
+        }
+
+        TEST_STR = builder.toString();
+    }
+
+    @Test
+    public void testCompression() throws UnsupportedEncodingException {
+        Compressor compressor = ExtensionLoaderFactory.getExtensionLoader(Compressor.class).getExtension("gzip");
+        Assert.assertTrue(compressor instanceof GzipRpcCompressor);
+
+        byte[] bs = compressor.compress(TEST_STR.getBytes("utf-8"));
+        String s1 = new String(compressor.deCompress(bs), "utf-8");
+        Assert.assertEquals(TEST_STR, s1);
+    }
+
+}