FEATURE: Add pluggable compression codec support with Snappy algorithm

pom.xml

@@ -20,6 +20,7 @@
         <bolyartech.version>2.0.2</bolyartech.version>
         <jackson.version>2.19.0</jackson.version>
         <findbugs-jsr305.version>3.0.2</findbugs-jsr305.version>
+        <snappy-java.version>1.1.10.8</snappy-java.version>
     </properties>
     <licenses>
         <license>
@@ -159,6 +160,14 @@
             <version>${findbugs-jsr305.version}</version>
             <scope>provided</scope>
         </dependency>
+
+        <!-- Snappy -->
+        <dependency>
+            <groupId>org.xerial.snappy</groupId>
+            <artifactId>snappy-java</artifactId>
+            <version>${snappy-java.version}</version>
+            <scope>compile</scope>
+        </dependency>
     </dependencies>
 
     <build>

src/main/java/net/spy/memcached/transcoders/BaseSerializingTranscoder.java

@@ -26,6 +26,8 @@
 import java.lang.reflect.Proxy;
 
 import net.spy.memcached.compat.SpyObject;
+import net.spy.memcached.transcoders.compression.CompressionCodecIF;
+import net.spy.memcached.transcoders.compression.GZIPCompressionCodec;
 
 /**
  * Base class for any transcoders that may want to work with serialized or
@@ -42,7 +44,7 @@ public abstract class BaseSerializingTranscoder extends SpyObject {
    */
   private final ClassLoader classLoader;
 
-  private final CompressionUtils cu = new CompressionUtils();
+  private final CompressionCodecIF compressionCodec;
   protected final TranscoderUtils tu;
 
   /**
@@ -57,9 +59,15 @@ public BaseSerializingTranscoder(int max, ClassLoader cl) {
   }
 
   public BaseSerializingTranscoder(int max, ClassLoader cl, boolean pack) {
+    this(max, cl, pack, new GZIPCompressionCodec());
+  }
+
+  public BaseSerializingTranscoder(int max, ClassLoader cl, boolean pack,
+                                   CompressionCodecIF codec) {
     super();
     this.maxSize = max;
     this.classLoader = cl;
+    this.compressionCodec = codec;
     this.tu = new TranscoderUtils(pack);
   }
 
@@ -71,7 +79,7 @@ public BaseSerializingTranscoder(int max, ClassLoader cl, boolean pack) {
    * @param threshold the number of bytes
    */
   public void setCompressionThreshold(int threshold) {
-    cu.setCompressionThreshold(threshold);
+    compressionCodec.setCompressionThreshold(threshold);
   }
 
   public String getCharset() {
@@ -127,7 +135,7 @@ protected Object deserialize(byte[] in) {
    * @throws NullPointerException if the input is null
    */
   protected byte[] compress(byte[] in) {
-    return cu.compress(in);
+    return compressionCodec.compress(in);
   }
 
   /**
@@ -137,7 +145,7 @@ protected byte[] compress(byte[] in) {
    * @return the decompressed byte array, or null if input is null or decompression fails
    */
   protected byte[] decompress(byte[] in) {
-    return cu.decompress(in);
+    return compressionCodec.decompress(in);
   }
 
   /**
@@ -147,7 +155,7 @@ protected byte[] decompress(byte[] in) {
    * @return true if the data should be compressed, false otherwise
    */
   protected boolean isCompressionCandidate(byte[] data) {
-    return cu.isCompressionCandidate(data);
+    return compressionCodec.isCompressionCandidate(data);
   }
 
   public int getMaxSize() {

src/main/java/net/spy/memcached/transcoders/CompressionUtils.java

@@ -12,6 +12,7 @@
 /**
  * Utility class for compression and decompression operations.
  */
+@Deprecated
 public class CompressionUtils extends SpyObject {
 
   public static final int DEFAULT_COMPRESSION_THRESHOLD = 16384;

src/main/java/net/spy/memcached/transcoders/GenericJsonSerializingTranscoder.java

@@ -10,6 +10,8 @@
 
 import net.spy.memcached.CachedData;
 import net.spy.memcached.compat.SpyObject;
+import net.spy.memcached.transcoders.compression.CompressionCodecIF;
+import net.spy.memcached.transcoders.compression.GZIPCompressionCodec;
 
 import static net.spy.memcached.transcoders.TranscoderUtils.COMPRESSED;
 import static net.spy.memcached.transcoders.TranscoderUtils.SERIALIZED;
@@ -30,7 +32,7 @@ public class GenericJsonSerializingTranscoder extends SpyObject implements Trans
 
   private final ObjectMapper objectMapper;
   private final int maxSize;
-  private final CompressionUtils cu;
+  private final CompressionCodecIF compressionCodec;
   private final TranscoderUtils tu;
   private final boolean isCollection;
   private final boolean forceJsonSerializeForCollection;
@@ -53,15 +55,7 @@ public GenericJsonSerializingTranscoder(ObjectMapper objectMapper, String typeHi
 
   @Deprecated
   public GenericJsonSerializingTranscoder(ObjectMapper objectMapper, int max) {
-    if (objectMapper == null) {
-      throw new IllegalArgumentException("ObjectMapper must not be null");
-    }
-    this.objectMapper = objectMapper;
-    this.maxSize = max;
-    this.cu = new CompressionUtils();
-    this.tu = new TranscoderUtils(true);
-    this.isCollection = false;
-    this.forceJsonSerializeForCollection = false;
+    this(objectMapper, max, false, false, new GZIPCompressionCodec());
   }
 
   /**
@@ -71,13 +65,14 @@ public GenericJsonSerializingTranscoder(ObjectMapper objectMapper, int max) {
    */
   private GenericJsonSerializingTranscoder(ObjectMapper objectMapper, int max,
                                            boolean isCollection,
-                                           boolean forceJsonSerializeForCollection) {
+                                           boolean forceJsonSerializeForCollection,
+                                           CompressionCodecIF codec) {
     if (objectMapper == null) {
       throw new IllegalArgumentException("ObjectMapper must not be null");
     }
     this.objectMapper = objectMapper;
     this.maxSize = max;
-    this.cu = new CompressionUtils();
+    this.compressionCodec = codec;
     this.tu = new TranscoderUtils(true);
     this.isCollection = isCollection;
     this.forceJsonSerializeForCollection = forceJsonSerializeForCollection;
@@ -131,7 +126,7 @@ public boolean isForceSerializeForCollection() {
    * @param threshold the number of bytes
    */
   public void setCompressionThreshold(int threshold) {
-    cu.setCompressionThreshold(threshold);
+    compressionCodec.setCompressionThreshold(threshold);
   }
 
   /**
@@ -153,7 +148,7 @@ public Object decode(CachedData d) {
     }
 
     if ((d.getFlags() & COMPRESSED) != 0) {
-      data = cu.decompress(data);
+      data = compressionCodec.decompress(data);
     }
 
     Object rv = null;
@@ -241,8 +236,8 @@ public CachedData encode(Object o) {
       flags |= SERIALIZED;
     }
     assert b != null;
-    if (!isCollection && cu.isCompressionCandidate(b)) {
-      byte[] compressed = cu.compress(b);
+    if (!isCollection && compressionCodec.isCompressionCandidate(b)) {
+      byte[] compressed = compressionCodec.compress(b);
       if (compressed.length < b.length) {
         getLogger().debug("Compressed %s from %d to %d",
             o.getClass().getName(), b.length, compressed.length);
@@ -287,6 +282,7 @@ public static final class Builder {
     private int max;
     private boolean isCollection;
     private boolean forceJsonSerializeForCollection;
+    private CompressionCodecIF compressionCodec = new GZIPCompressionCodec();
 
     private Builder(ObjectMapper objectMapper) {
       this.objectMapper = objectMapper;
@@ -326,6 +322,11 @@ public Builder typeHintPropertyName(String typeHintPropertyName) {
       return this;
     }
 
+    public Builder compressionCodec(CompressionCodecIF codec) {
+      this.compressionCodec = codec;
+      return this;
+    }
+
     public Builder forceJsonSerializeForCollection() {
       if (!isCollection) {
         throw new IllegalStateException("forceJsonSerializationForCollection can only be " +
@@ -346,7 +347,7 @@ public GenericJsonSerializingTranscoder build() {
         objectMapper.setDefaultTyping(typer);
       }
       return new GenericJsonSerializingTranscoder(objectMapper, max,
-          isCollection, forceJsonSerializeForCollection);
+          isCollection, forceJsonSerializeForCollection, compressionCodec);
     }
   }
 }

src/main/java/net/spy/memcached/transcoders/JsonSerializingTranscoder.java

@@ -26,6 +26,8 @@
 
 import net.spy.memcached.CachedData;
 import net.spy.memcached.compat.SpyObject;
+import net.spy.memcached.transcoders.compression.CompressionCodecIF;
+import net.spy.memcached.transcoders.compression.GZIPCompressionCodec;
 
 import static net.spy.memcached.transcoders.TranscoderUtils.COMPRESSED;
 import static net.spy.memcached.transcoders.TranscoderUtils.SERIALIZED;
@@ -48,7 +50,7 @@ public class JsonSerializingTranscoder<T> extends SpyObject implements Transcode
   private final ObjectMapper objectMapper;
   private final JavaType javaType;
   private final int maxSize;
-  private final CompressionUtils cu;
+  private final CompressionCodecIF compressionCodec;
   private final TranscoderUtils tu;
 
   public JsonSerializingTranscoder(Class<T> clazz) {
@@ -60,18 +62,26 @@ public JsonSerializingTranscoder(JavaType javaType) {
   }
 
   public JsonSerializingTranscoder(int max, Class<T> clazz) {
+    this(max, clazz, new GZIPCompressionCodec());
+  }
+
+  public JsonSerializingTranscoder(int max, JavaType javaType) {
+    this(max, javaType, new GZIPCompressionCodec());
+  }
+
+  public JsonSerializingTranscoder(int max, Class<T> clazz, CompressionCodecIF codec) {
     this.maxSize = max;
     this.objectMapper = new ObjectMapper();
     this.javaType = objectMapper.getTypeFactory().constructType(clazz);
-    this.cu = new CompressionUtils();
+    this.compressionCodec = codec;
     this.tu = new TranscoderUtils(true);
   }
 
-  public JsonSerializingTranscoder(int max, JavaType javaType) {
+  public JsonSerializingTranscoder(int max, JavaType javaType, CompressionCodecIF codec) {
     this.maxSize = max;
     this.objectMapper = new ObjectMapper();
     this.javaType = Objects.requireNonNull(javaType, "JavaType must not be null");
-    this.cu = new CompressionUtils();
+    this.compressionCodec = codec;
     this.tu = new TranscoderUtils(true);
   }
 
@@ -88,7 +98,7 @@ public int getMaxSize() {
    * @param threshold the number of bytes
    */
   public void setCompressionThreshold(int threshold) {
-    cu.setCompressionThreshold(threshold);
+    this.compressionCodec.setCompressionThreshold(threshold);
   }
 
   /**
@@ -110,7 +120,7 @@ public T decode(CachedData d) {
     }
 
     if ((d.getFlags() & COMPRESSED) != 0) {
-      data = cu.decompress(data);
+      data = compressionCodec.decompress(data);
     }
 
     Object rv = null;
@@ -192,8 +202,8 @@ public CachedData encode(T o) {
       flags |= SERIALIZED;
     }
     assert b != null;
-    if (cu.isCompressionCandidate(b)) {
-      byte[] compressed = cu.compress(b);
+    if (compressionCodec.isCompressionCandidate(b)) {
+      byte[] compressed = compressionCodec.compress(b);
       if (compressed.length < b.length) {
         getLogger().debug("Compressed %s from %d to %d",
                 o.getClass().getName(), b.length, compressed.length);

src/main/java/net/spy/memcached/transcoders/SerializingTranscoder.java

@@ -21,6 +21,8 @@
 import java.util.Date;
 
 import net.spy.memcached.CachedData;
+import net.spy.memcached.transcoders.compression.CompressionCodecIF;
+import net.spy.memcached.transcoders.compression.GZIPCompressionCodec;
 
 import static net.spy.memcached.transcoders.TranscoderUtils.COMPRESSED;
 import static net.spy.memcached.transcoders.TranscoderUtils.SERIALIZED;
@@ -54,15 +56,20 @@ public class SerializingTranscoder extends BaseSerializingTranscoder
    * Get a serializing transcoder with the default max data size.
    */
   public SerializingTranscoder() {
-    this(CachedData.MAX_SIZE, null, false, false);
+    this(CachedData.MAX_SIZE, null, false, false, new GZIPCompressionCodec());
   }
 
   public SerializingTranscoder(int max) {
-    this(max, null, false, false);
+    this(max, null, false, false, new GZIPCompressionCodec());
   }
 
   public SerializingTranscoder(int max, ClassLoader cl) {
-    this(max, cl, false, false);
+    this(max, cl, false, false, new GZIPCompressionCodec());
+  }
+
+  public SerializingTranscoder(int max, ClassLoader cl, boolean isCollection,
+                               boolean forceJDKSerializeForCollection) {
+    this(max, cl, isCollection, forceJDKSerializeForCollection, new GZIPCompressionCodec());
   }
 
   /**
@@ -71,8 +78,9 @@ public SerializingTranscoder(int max, ClassLoader cl) {
    * or Builder for custom configurations.
    */
   protected SerializingTranscoder(int max, ClassLoader cl, boolean isCollection,
-                               boolean forceJDKSerializeForCollection) {
-    super(max, cl);
+                               boolean forceJDKSerializeForCollection,
+                               CompressionCodecIF codec) {
+    super(max, cl, true, codec);
     this.isCollection = isCollection;
     this.forceJDKSerializeForCollection = forceJDKSerializeForCollection;
   }
@@ -207,6 +215,7 @@ public static final class Builder {
     private ClassLoader cl;
     private boolean isCollection;
     private boolean forceJDKSerializeForCollection;
+    private CompressionCodecIF compressionCodec = new GZIPCompressionCodec();
 
     private Builder() {}
 
@@ -236,6 +245,11 @@ public Builder classLoader(ClassLoader cl) {
       return this;
     }
 
+    public Builder compressionCodec(CompressionCodecIF codec) {
+      this.compressionCodec = codec;
+      return this;
+    }
+
     /**
      * By default, this transcoder uses Java serialization only if the type is a user-defined class.
      * This mechanism may cause malfunction if you store Object type values
@@ -255,7 +269,8 @@ public Builder forceJDKSerializationForCollection() {
     }
 
     public SerializingTranscoder build() {
-      return new SerializingTranscoder(max, cl, isCollection, forceJDKSerializeForCollection);
+      return new SerializingTranscoder(max, cl, isCollection,
+              forceJDKSerializeForCollection, compressionCodec);
     }
   }
 }

src/main/java/net/spy/memcached/transcoders/compression/CompressionCodecIF.java

@@ -0,0 +1,12 @@
+package net.spy.memcached.transcoders.compression;
+
+public interface CompressionCodecIF {
+
+  byte[] compress(byte[] data);
+
+  byte[] decompress(byte[] data);
+
+  boolean isCompressionCandidate(byte[] data);
+
+  void setCompressionThreshold(int threshold);
+}