MessagePackRecordFormatter extends AbstractRecordFormatter { @Override public byte[] formatFromMessagePack(String tag, Object timestamp, byte[] mapValue, int offset, int len) { try { MessagePackRecordAccessor recordAccessor = new MessagePackRecordAccessor(ByteBuffer.wrap(mapValue, offset, len)); recordAccessor.setTimestamp(getEpoch(timestamp)); return recordAccessor.toMessagePack(objectMapperForMessagePack); } catch (Throwable e) { LOG.error(String.format( "Failed to format a MessagePack record: cause=%s, tag=%s, timestamp=%s, offset=%d, len=%d", e.getMessage(), tag, timestamp, offset, len)); throw e; } } MessagePackRecordFormatter();  MessagePackRecordFormatter(Config config); @Override byte[] format(String tag, Object timestamp, Map<String, Object> data); @Override byte[] formatFromMessagePack(String tag, Object timestamp, byte[] mapValue, int offset, int len); @Override byte[] formatFromMessagePack(String tag, Object timestamp, ByteBuffer mapValue); @Override String formatName();  }
@Test void formatFromMessagePackBytes() throws IOException { long now = System.currentTimeMillis() / 1000; ObjectMapper objectMapper = new ObjectMapper(new MessagePackFactory()); { byte[] bytes = objectMapper.writeValueAsBytes(RECORD_0); assertRecord0(recordFormatter.formatFromMessagePack(TAG, now, bytes, 0, bytes.length), now); } { byte[] bytes = objectMapper.writeValueAsBytes(RECORD_1); assertRecord1(recordFormatter.formatFromMessagePack(TAG, now, bytes, 0, bytes.length), FUTURE_EPOCH); } { byte[] bytes = objectMapper.writeValueAsBytes(RECORD_2); assertRecord2(recordFormatter.formatFromMessagePack(TAG, now, bytes, 0, bytes.length), now); } }
@Test void formatFromMessagePackByteBuffer() throws IOException { long now = System.currentTimeMillis() / 1000; assertRecord0(recordFormatter.formatFromMessagePack(TAG, now, convertMapToMessagePackByteBuffer(RECORD_0, false)), now); assertRecord1(recordFormatter.formatFromMessagePack(TAG, now, convertMapToMessagePackByteBuffer(RECORD_1, true)), FUTURE_EPOCH); assertRecord2(recordFormatter.formatFromMessagePack(TAG, now, convertMapToMessagePackByteBuffer(RECORD_2, true)), now); }