BinaryMemcacheClientCodec Class — netty Architecture

Source Code

codec-memcache/src/main/java/io/netty/handler/codec/memcache/binary/BinaryMemcacheClientCodec.java lines 39–121

@UnstableApi
public final class BinaryMemcacheClientCodec extends
        CombinedChannelDuplexHandler<BinaryMemcacheResponseDecoder, BinaryMemcacheRequestEncoder> {

    private final boolean failOnMissingResponse;
    private final AtomicLong requestResponseCounter = new AtomicLong();

    /**
     * Create a new {@link BinaryMemcacheClientCodec} with the default settings applied.
     */
    public BinaryMemcacheClientCodec() {
        this(AbstractBinaryMemcacheDecoder.DEFAULT_MAX_CHUNK_SIZE);
    }

    /**
     * Create a new {@link BinaryMemcacheClientCodec} and set a custom chunk size.
     *
     * @param decodeChunkSize the maximum chunk size.
     */
    public BinaryMemcacheClientCodec(int decodeChunkSize) {
        this(decodeChunkSize, false);
    }

    /**
     * Create a new {@link BinaryMemcacheClientCodec} with custom settings.
     *
     * @param decodeChunkSize       the maximum chunk size.
     * @param failOnMissingResponse report if after close there are outstanding requests.
     */
    public BinaryMemcacheClientCodec(int decodeChunkSize, boolean failOnMissingResponse) {
        this.failOnMissingResponse = failOnMissingResponse;
        init(new Decoder(decodeChunkSize), new Encoder());
    }

    private final class Encoder extends BinaryMemcacheRequestEncoder {

        @Override
        protected void encode(ChannelHandlerContext ctx, Object msg, List<Object> out) throws Exception {
            super.encode(ctx, msg, out);

            if (failOnMissingResponse && msg instanceof LastMemcacheContent) {
                requestResponseCounter.incrementAndGet();
            }
        }
    }

    private final class Decoder extends BinaryMemcacheResponseDecoder {

        Decoder(int chunkSize) {
            super(chunkSize);
        }

        @Override
        protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception {
            int oldSize = out.size();
            super.decode(ctx, in, out);

            if (failOnMissingResponse) {
                final int size = out.size();
                for (int i = oldSize; i < size; i ++) {
                    Object msg = out.get(i);
                    if (msg instanceof LastMemcacheContent) {
                        requestResponseCounter.decrementAndGet();
                    }
                }
            }
        }

        @Override
        public void channelInactive(ChannelHandlerContext ctx) throws Exception {
            super.channelInactive(ctx);

            if (failOnMissingResponse) {
                long missingResponses = requestResponseCounter.get();
                if (missingResponses > 0) {
                    ctx.fireExceptionCaught(new PrematureChannelClosureException(
                        "channel gone inactive with " + missingResponses +
                            " missing response(s)"));
                }
            }
        }