FluentdIngester implements Ingester { @Override public void ingest(String tag, ByteBuffer dataBuffer) throws IOException { ByteArrayOutputStream header = new ByteArrayOutputStream(); MessagePacker messagePacker = MessagePack.newDefaultPacker(header); int dataLength = dataBuffer.limit(); messagePacker.packArrayHeader(3); messagePacker.packString(tag); messagePacker.packRawStringHeader(dataLength); messagePacker.flush(); ByteBuffer headerBuffer = ByteBuffer.wrap(header.toByteArray()); if (config.isAckResponseMode()) { byte[] uuidBytes = UUID.randomUUID().toString().getBytes(CHARSET); ByteBuffer optionBuffer = ByteBuffer.wrap(objectMapper.writeValueAsBytes(new RequestOption(dataLength, uuidBytes))); List<ByteBuffer> buffers = Arrays.asList(headerBuffer, dataBuffer, optionBuffer); synchronized (sender) { sender.sendWithAck(buffers, uuidBytes); } } else { ByteBuffer optionBuffer = ByteBuffer.wrap(objectMapper.writeValueAsBytes(new RequestOption(dataLength, null))); List<ByteBuffer> buffers = Arrays.asList(headerBuffer, dataBuffer, optionBuffer); synchronized (sender) { sender.send(buffers); } } } FluentdIngester(FluentdSender sender);  FluentdIngester(Config config, FluentdSender sender); @Override void ingest(String tag, ByteBuffer dataBuffer); @Override Sender getSender(); boolean isAckResponseMode(); @Override void close();  }
@Test void ingestWithoutAck() throws IOException { Ingester ingester = new FluentdIngester(new FluentdIngester.Config(), fluentdSender); ingester.ingest(TAG, ByteBuffer.wrap(DATA)); verify(fluentdSender, times(1)).send(byteBuffersArgumentCaptor.capture()); List<ByteBuffer> byteBuffers = byteBuffersArgumentCaptor.getAllValues().get(0); byte[] ingested = getIngestedData(byteBuffers); MessageUnpacker unpacker = MessagePack.newDefaultUnpacker(ingested); ImmutableArrayValue arrayValue = unpacker.unpackValue().asArrayValue(); assertEquals(3, arrayValue.size()); assertEquals(TAG, arrayValue.get(0).asStringValue().asString()); assertArrayEquals(DATA, arrayValue.get(1).asRawValue().asByteArray()); Map<Value, Value> options = arrayValue.get(2).asMapValue().map(); assertEquals(1, options.size()); assertEquals(DATA.length, options.get(ValueFactory.newString("size")).asIntegerValue().asInt()); }
@Test void ingestWithAck() throws IOException { FluentdIngester.Config config = new FluentdIngester.Config(); config.setAckResponseMode(true); Ingester ingester = new FluentdIngester(config, fluentdSender); ingester.ingest(TAG, ByteBuffer.wrap(DATA)); ArgumentCaptor<byte[]> ackTokenArgumentCaptor = ArgumentCaptor.forClass(byte[].class); verify(fluentdSender, times(1)) .sendWithAck(byteBuffersArgumentCaptor.capture(), ackTokenArgumentCaptor.capture()); List<ByteBuffer> byteBuffers = byteBuffersArgumentCaptor.getAllValues().get(0); byte[] ingested = getIngestedData(byteBuffers); MessageUnpacker unpacker = MessagePack.newDefaultUnpacker(ingested); ImmutableArrayValue arrayValue = unpacker.unpackValue().asArrayValue(); assertEquals(3, arrayValue.size()); assertEquals(TAG, arrayValue.get(0).asStringValue().asString()); assertArrayEquals(DATA, arrayValue.get(1).asRawValue().asByteArray()); Map<Value, Value> options = arrayValue.get(2).asMapValue().map(); assertEquals(2, options.size()); assertEquals(DATA.length, options.get(ValueFactory.newString("size")).asIntegerValue().asInt()); String ackToken = options.get(ValueFactory.newString("chunk")).asRawValue().asString(); UUID uuidFromAckToken = UUID.fromString(ackToken); List<byte[]> ackTokenArgumentCaptorAllValues = ackTokenArgumentCaptor.getAllValues(); assertEquals(1, ackTokenArgumentCaptorAllValues.size()); assertEquals(uuidFromAckToken, UUID.fromString(new String(ackTokenArgumentCaptorAllValues.get(0), CHARSET))); }