@Nonnull
    @Override
    public Result addChunk(ByteBuf buffer) {
        final byte[] readable = new byte[buffer.readableBytes()];
        buffer.readBytes(readable, buffer.readerIndex(), buffer.readableBytes());

        final GELFMessage msg = new GELFMessage(readable);

        final ByteBuf aggregatedBuffer;
        switch (msg.getGELFType()) {
            case CHUNKED:
                try {
                    chunkCounter.inc();
                    aggregatedBuffer = checkForCompletion(msg);
                    if (aggregatedBuffer == null) {
                        return VALID_EMPTY_RESULT;
                    }
                } catch (IllegalArgumentException | IllegalStateException | IndexOutOfBoundsException e) {
                    log.debug("Invalid gelf message chunk, dropping message.", e);
                    return INVALID_RESULT;
                }
                break;
            case ZLIB:
            case GZIP:
            case UNCOMPRESSED:
                aggregatedBuffer = Unpooled.wrappedBuffer(readable);
                break;
            case UNSUPPORTED:
                return INVALID_RESULT;
            default:
                return INVALID_RESULT;
        }
        return new Result(aggregatedBuffer, true);
    }
@Test
    public void duplicateChunk() {
        final byte[] messageId1 = generateMessageId(1);
        final byte[] messageId2 = generateMessageId(2);
        final ByteBuf chunk1 = createChunk(messageId1, (byte) 0, (byte) 2, new byte[16]);
        final ByteBuf chunk2 = createChunk(messageId1, (byte) 0, (byte) 2, new byte[16]);
        final ByteBuf chunk3 = createChunk(messageId2, (byte) 0, (byte) 2, new byte[16]);
        final ByteBuf chunk4 = createChunk(messageId1, (byte) 1, (byte) 2, new byte[16]);
        final ByteBuf chunk5 = createChunk(messageId2, (byte) 1, (byte) 2, new byte[16]);

        assertNull("message should not be complete", aggregator.addChunk(chunk1).getMessage());
        assertNull("message should not be complete", aggregator.addChunk(chunk2).getMessage());
        assertNull("message should not be complete", aggregator.addChunk(chunk3).getMessage());
        assertNotNull("message 1 should be complete", aggregator.addChunk(chunk4).getMessage());
        assertNotNull("message 2 should be complete", aggregator.addChunk(chunk5).getMessage());

        assertEquals(2, counterValueNamed(metricRegistry, COMPLETE_MESSAGES));
        assertEquals(5, counterValueNamed(metricRegistry, CHUNK_COUNTER));
        assertEquals(0, counterValueNamed(metricRegistry, WAITING_MESSAGES));
        assertEquals(0, counterValueNamed(metricRegistry, EXPIRED_CHUNKS));
        assertEquals(0, counterValueNamed(metricRegistry, EXPIRED_MESSAGES));
        assertEquals(1, counterValueNamed(metricRegistry, DUPLICATE_CHUNKS));
    }