@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 differentIdsDoNotInterfere() {
        final ByteBuf[] msg1 = createChunkedMessage(4096 + 1, 1024, generateMessageId(1));// 5 chunks;
        final ByteBuf[] msg2 = createChunkedMessage(4096 + 1, 1024, generateMessageId(2));// 5 chunks;

        CodecAggregator.Result result1 = null;
        CodecAggregator.Result result2 = null;
        for (int i = 0; i < msg1.length; i++) {
            result1 = aggregator.addChunk(msg1[i]);
            if (i > 0) {
                result2 = aggregator.addChunk(msg2[i]);
            }
        }
        assertNotNull(result1);
        assertNotNull(result2);
        assertNotNull("message 1 should be complete", result1.getMessage());
        assertNull("message 2 should not be complete", result2.getMessage());
        // only one is complete, we sent 9 chunks
        assertEquals(1, counterValueNamed(metricRegistry, COMPLETE_MESSAGES));
        assertEquals(9, counterValueNamed(metricRegistry, CHUNK_COUNTER));
        assertEquals(1, counterValueNamed(metricRegistry, WAITING_MESSAGES));
        assertEquals(0, counterValueNamed(metricRegistry, EXPIRED_CHUNKS));
        assertEquals(0, counterValueNamed(metricRegistry, EXPIRED_MESSAGES));
        assertEquals(0, counterValueNamed(metricRegistry, DUPLICATE_CHUNKS));
    }