MetricsBufferDecoder extends ByteToMessageDecoder { @Override protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception { if (in.readableBytes() < 8) { return; } LOG.debug("Received {} bytes", in.readableBytes()); boolean isFlatBuffer = false; try { isFlatBuffer = Metrics.MetricsBufferHasIdentifier(in.slice().nioBuffer()); } catch (Exception e) { LOG.warn("Error checking for identifier: " + e.getClass().getSimpleName()); isFlatBuffer = false; } if (!isFlatBuffer) { LOG.debug("Input is not a Flatbuffer"); out.add(in.readBytes(in.readableBytes())); } else { LOG.debug("Input is a Flatbuffer"); ByteBuf copy = in.readBytes(in.readableBytes()); try { Metrics metrics = Metrics.getRootAsMetrics(copy.nioBuffer()); int length = metrics.metricsLength(); for (int i = 0; i < length; i++) { MetricRequest m = new MetricRequest(parseFlatbuffer(metrics.metrics(i))); LOG.debug("Returning {}", m); out.add(m); } } catch (Exception e) { LOG.warn("Error decoding byte[] with Google Flatbuffers: " + e.getMessage()); return; } finally { copy.release(); } } }  Metric parseFlatbuffer(timely.api.flatbuffer.Metric flatMetric);  }
@Test public void testBuffer() throws Exception { FlatBufferBuilder builder = new FlatBufferBuilder(1); int[] metric = new int[1]; Map<String, String> t = new HashMap<>(); t.put("tag1", "value1"); t.put("tag2", "value2"); metric[0] = createMetric(builder, "sys.cpu.user", TEST_TIME, 1.0D, t); int metricVector = timely.api.flatbuffer.Metrics.createMetricsVector(builder, metric); timely.api.flatbuffer.Metrics.startMetrics(builder); timely.api.flatbuffer.Metrics.addMetrics(builder, metricVector); int metrics = timely.api.flatbuffer.Metrics.endMetrics(builder); timely.api.flatbuffer.Metrics.finishMetricsBuffer(builder, metrics); ByteBuf buf = Unpooled.wrappedBuffer(builder.dataBuffer()); MetricsBufferDecoder decoder = new MetricsBufferDecoder(); List<Object> results = new ArrayList<Object>(); decoder.decode(null, buf, results); Assert.assertEquals(1, results.size()); MetricRequest expected = new MetricRequest( Metric.newBuilder() .name("sys.cpu.user") .value(TEST_TIME, 1.0D) .tag(new Tag("tag1", "value1")) .tag(new Tag("tag2", "value2")) .build() ); Assert.assertEquals(expected, results.get(0)); }
@Test public void testPutMultipleBuffers() throws Exception { FlatBufferBuilder builder = new FlatBufferBuilder(1); int[] metric = new int[2]; Map<String, String> t = new HashMap<>(); t.put("tag1", "value1"); t.put("tag2", "value2"); metric[0] = createMetric(builder, "sys.cpu.user", TEST_TIME, 1.0D, t); t = new HashMap<>(); t.put("tag3", "value3"); t.put("tag4", "value4"); metric[1] = createMetric(builder, "sys.cpu.idle", TEST_TIME, 1.0D, t); int metricVector = timely.api.flatbuffer.Metrics.createMetricsVector(builder, metric); timely.api.flatbuffer.Metrics.startMetrics(builder); timely.api.flatbuffer.Metrics.addMetrics(builder, metricVector); int metrics = timely.api.flatbuffer.Metrics.endMetrics(builder); timely.api.flatbuffer.Metrics.finishMetricsBuffer(builder, metrics); ByteBuf buf = Unpooled.wrappedBuffer(builder.dataBuffer()); MetricsBufferDecoder decoder = new MetricsBufferDecoder(); List<Object> results = new ArrayList<Object>(); decoder.decode(null, buf, results); Assert.assertEquals(2, results.size()); MetricRequest expected = new MetricRequest( Metric.newBuilder() .name("sys.cpu.user") .value(TEST_TIME, 1.0D) .tag(new Tag("tag1", "value1")) .tag(new Tag("tag2", "value2")) .build() ); Assert.assertEquals(expected, results.get(0)); expected = new MetricRequest( Metric.newBuilder() .name("sys.cpu.idle") .value(TEST_TIME, 1.0D) .tag(new Tag("tag3", "value3")) .tag(new Tag("tag4", "value4")) .build() ); Assert.assertEquals(expected, results.get(1)); }