ZstdDecoder Class — netty Architecture

Source Code

codec-compression/src/main/java/io/netty/handler/codec/compression/ZstdDecoder.java lines 34–189

public final class ZstdDecoder extends ByteToMessageDecoder {
    // Don't use static here as we want to still allow to load the classes.
    {
        try {
            Zstd.ensureAvailability();
        } catch (Throwable throwable) {
            throw new ExceptionInInitializerError(throwable);
        }
    }

    private final int maximumAllocationSize;
    private final MutableByteBufInputStream inputStream = new MutableByteBufInputStream();
    private ZstdInputStreamNoFinalizer zstdIs;

    private boolean needsRead;
    private State currentState = State.DECOMPRESS_DATA;

    /**
     * Current state of stream.
     */
    private enum State {
        DECOMPRESS_DATA,
        CORRUPTED
    }

    public ZstdDecoder() {
        this(4 * 1024 * 1024);
    }

    public ZstdDecoder(int maximumAllocationSize) {
        this.maximumAllocationSize = ObjectUtil.checkPositiveOrZero(maximumAllocationSize, "maximumAllocationSize");
    }

    @Override
    protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception {
        needsRead = true;
        try {
            if (currentState == State.CORRUPTED) {
                in.skipBytes(in.readableBytes());

                return;
            }
            inputStream.current = in;

            ByteBuf outBuffer = null;

            final int compressedLength = in.readableBytes();
            try {
                long uncompressedLength;
                if (in.isDirect()) {
                    uncompressedLength = com.github.luben.zstd.Zstd.getFrameContentSize(
                            CompressionUtil.safeNioBuffer(in, in.readerIndex(), in.readableBytes()));
                } else {
                    uncompressedLength = com.github.luben.zstd.Zstd.getFrameContentSize(
                            in.array(), in.readerIndex() + in.arrayOffset(), in.readableBytes());
                }
                if (uncompressedLength <= 0) {
                    // Let's start with the compressedLength * 2 as often we will not have everything
                    // we need in the in buffer and don't want to reserve too much memory.
                    uncompressedLength = compressedLength * 2L;
                }

                int w;
                do {
                    if (outBuffer == null) {
                        outBuffer = ctx.alloc().heapBuffer((int) (maximumAllocationSize == 0 ?
                                uncompressedLength : Math.min(maximumAllocationSize, uncompressedLength)));
                    }
                    do {
                        w = outBuffer.writeBytes(zstdIs, outBuffer.writableBytes());
                    } while (w != -1 && outBuffer.isWritable());
                    if (outBuffer.isReadable()) {
                        needsRead = false;
                        ctx.fireChannelRead(outBuffer);
                        outBuffer = null;
                    }
                } while (w != -1);
            } finally {
                if (outBuffer != null) {
                    outBuffer.release();
                }