Java Snappy compressor wrapper for a lettuce codec to store snappy-compressed values within Redis

SnappyCompressor.java

import static com.google.common.base.Preconditions.checkArgument;

import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.nio.ByteBuffer;

import org.xerial.snappy.SnappyInputStream;
import org.xerial.snappy.SnappyOutputStream;

import com.google.common.io.ByteStreams;
import com.lambdaworks.redis.codec.RedisCodec;

import io.netty.buffer.ByteBufInputStream;
import io.netty.buffer.Unpooled;

/**
 * Compression wrapper for a {@link RedisCodec} to apply {@link org.xerial.snappy.Snappy} compression on values. The wrapper is
 * used by calling {@link #wrap(RedisCodec)} with the data type codec such as byte[] or UTF8.
 */
public class SnappyCompressor implements RedisCodec<Object, Object> {

    private RedisCodec<Object, Object> delegate;

    /**
     * A {@link RedisCodec} that compresses values from a delegating {@link RedisCodec} using snappy compression.
     * 
     * @param delegate codec used for key-value encoding/decoding, must not be {@literal null}.
     * @param <K> Key type.
     * @param <V> Value type.
     * @return Value-compressing codec.
     */
    @SuppressWarnings({ "rawtypes", "unchecked" })
    public static <K, V> RedisCodec<K, V> wrap(RedisCodec<K, V> delegate) {
        checkArgument(delegate != null, "RedisCodec delegate must not be null");
        return (RedisCodec) new SnappyCompressor((RedisCodec) delegate);
    }

    private SnappyCompressor(RedisCodec<Object, Object> delegate) {
        this.delegate = delegate;
    }

    @Override
    public Object decodeKey(ByteBuffer bytes) {
        return delegate.decodeKey(bytes);
    }

    @Override
    public Object decodeValue(ByteBuffer bytes) {
        try {
            return delegate.decodeValue(decompress(bytes));
        } catch (IOException e) {
            throw new IllegalStateException(e);
        }
    }

    @Override
    public ByteBuffer encodeKey(Object key) {
        return delegate.encodeKey(key);
    }

    @Override
    public ByteBuffer encodeValue(Object value) {
        try {
            return compress(delegate.encodeValue(value));
        } catch (IOException e) {
            throw new IllegalStateException(e);
        }
    }

    private ByteBuffer compress(ByteBuffer source) throws IOException {
        if (source.remaining() == 0) {
            return source;
        }

        ByteArrayOutputStream buffer = new ByteArrayOutputStream();
        SnappyOutputStream compressor = new SnappyOutputStream(buffer);
        try {
            ByteStreams.copy(getByteBufInputStream(source), compressor);
        } finally {
            compressor.close();
        }
        return ByteBuffer.wrap(buffer.toByteArray());
    }

    private ByteBuffer decompress(ByteBuffer source) throws IOException {
        if (source.remaining() == 0) {
            return source;
        }

        ByteArrayOutputStream buffer = new ByteArrayOutputStream();
        SnappyInputStream decompressor = new SnappyInputStream(getByteBufInputStream(source));
        try {
            ByteStreams.copy(decompressor, buffer);
        } finally {
            decompressor.close();
        }
        return ByteBuffer.wrap(buffer.toByteArray());
    }

    private ByteBufInputStream getByteBufInputStream(ByteBuffer source) {
        return new ByteBufInputStream(Unpooled.wrappedBuffer(source));
    }
}

SnappyTest.java

import java.io.IOException;
import java.nio.ByteBuffer;

import com.lambdaworks.redis.*;
import com.lambdaworks.redis.api.sync.RedisCommands;
import org.junit.Test;

/**
 * @author <a href="mailto:mpaluch@paluch.biz">Mark Paluch</a>
 */
public class SnappyTest {

    @Test
    public void snappyTest() throws Exception {

        RedisClient redisClient = new RedisClient("localhost", 6379);
        RedisCommands<String, String> redisCommands = redisClient.connect(SnappyCompressor.wrap(new Utf8StringCodec())).sync();
        
        redisCommands.set("key", "My data is snappy-compressed. What about yours?");

        System.out.println(redisCommands.get("key")); // prints: My data is snappy-compressed. What about yours?

        redisCommands.close();
        redisClient.shutdown();
    }
}