class CompressionCodec {

    private CompressionCodec() {
    }

    /**
     * A {@link RedisCodec} that compresses values from a delegating {@link RedisCodec}.
     *
     * @param delegate codec used for key-value encoding/decoding, must not be {@code null}.
     * @param compressionType the compression type, must not be {@code null}.
     * @param <K> Key type.
     * @param <V> Value type.
     * @return Value-compressing codec.
     */
    @SuppressWarnings({ "rawtypes", "unchecked" })
    public static <K, V> RedisCodec<K, V> valueCompressor(RedisCodec<K, V> delegate, CompressionType compressionType) {<FILL_FUNCTION_BODY>}

    private static class CompressingValueCodecWrapper implements RedisCodec<Object, Object> {

        private RedisCodec<Object, Object> delegate;

        private CompressionType compressionType;

        public CompressingValueCodecWrapper(RedisCodec<Object, Object> delegate, CompressionType compressionType) {
            this.delegate = delegate;
            this.compressionType = compressionType;
        }

        @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 outputStream = new ByteArrayOutputStream(source.remaining() / 2);
            OutputStream compressor = null;

            try {
                try (ByteBufferInputStream sourceStream = new ByteBufferInputStream(source)) {
                    if (compressionType == CompressionType.GZIP) {
                        compressor = new GZIPOutputStream(outputStream);
                    }

                    if (compressionType == CompressionType.DEFLATE) {
                        compressor = new DeflaterOutputStream(outputStream);
                    }
                    copy(sourceStream, compressor);
                } finally {

                    if (compressor != null) {
                        compressor.close();
                    }
                }

                return ByteBuffer.wrap(outputStream.toByteArray());
            } finally {
                outputStream.close();
            }
        }

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

            InputStream decompressor = null;
            ByteArrayOutputStream outputStream = new ByteArrayOutputStream(source.remaining() * 2);

            try {
                try (ByteBufferInputStream sourceStream = new ByteBufferInputStream(source);) {

                    if (compressionType == CompressionType.GZIP) {
                        decompressor = new GZIPInputStream(sourceStream);
                    }

                    if (compressionType == CompressionType.DEFLATE) {
                        decompressor = new InflaterInputStream(sourceStream);
                    }

                    copy(decompressor, outputStream);
                } finally {
                    if (decompressor != null) {
                        decompressor.close();
                    }
                }

                return ByteBuffer.wrap(outputStream.toByteArray());
            } finally {
                outputStream.close();
            }
        }

    }

    /**
     * Copies all bytes from the input stream to the output stream. Does not close or flush either stream.
     *
     * @param from the input stream to read from
     * @param to the output stream to write to
     * @return the number of bytes copied
     * @throws IOException if an I/O error occurs
     */
    private static long copy(InputStream from, OutputStream to) throws IOException {
        LettuceAssert.notNull(from, "From must not be null");
        LettuceAssert.notNull(to, "From must not be null");
        byte[] buf = new byte[4096];
        long total = 0;
        while (true) {
            int r = from.read(buf);
            if (r == -1) {
                break;
            }
            to.write(buf, 0, r);
            total += r;
        }
        return total;
    }

    public enum CompressionType {
        GZIP, DEFLATE;
    }

}

        LettuceAssert.notNull(delegate, "RedisCodec must not be null");
        LettuceAssert.notNull(compressionType, "CompressionType must not be null");
        return (RedisCodec) new CompressingValueCodecWrapper((RedisCodec) delegate, compressionType);
    
<no_super_class>