TcpDecoder extends ByteToMessageDecoder { @Override protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception { ByteBuf buf = in.readBytes(in.readableBytes()); String input = ""; try { if (buf == Unpooled.EMPTY_BUFFER) { return; } if (buf.hasArray()) { input = new String(buf.array(), UTF_8); } else { input = buf.toString(UTF_8); } if (StringUtils.isEmpty(input)) { LOG.warn("Received no input"); return; } LOG.trace("Received input: {}", input); String[] parts = input.split(" "); String operation; if (parts.length == 0 && !StringUtils.isEmpty(input)) { operation = input; } else { operation = parts[0]; } TcpRequest tcp = null; try { tcp = AnnotationResolver.getClassForTcpOperation(operation); } catch (Exception e) { LOG.error("Error getting class for operation: " + operation, e); } if (null == tcp) { LOG.error("Unknown tcp operation:[{}] in input:[{}]", parts[0], input); return; } tcp.parse(input); out.add(tcp); LOG.trace("Converted {} to {}", input, tcp); } catch (Exception e) { LOG.error("{} parsing line:[{}]", e.getMessage(), input); } finally { buf.release(); } }    }
@Test public void testPutNoViz() throws Exception { TcpDecoder decoder = new TcpDecoder(); List<Object> results = new ArrayList<>(); String put = "put sys.cpu.user " + TEST_TIME + " 1.0 tag1=value1 tag2=value2"; ByteBuf buf = Unpooled.wrappedBuffer(put.getBytes()); decoder.decode(null, buf, results); Assert.assertEquals(1, results.size()); Assert.assertEquals(MetricRequest.class, results.get(0).getClass()); Metric m = ((MetricRequest) results.get(0)).getMetric(); Metric expected = Metric.newBuilder() .name("sys.cpu.user") .value(TEST_TIME, 1.0D) .tag(new Tag("tag1", "value1")) .tag(new Tag("tag2", "value2")) .build(); Assert.assertTrue(expected.equals(m)); }
@Test public void testPutWithViz() throws Exception { TcpDecoder decoder = new TcpDecoder(); List<Object> results = new ArrayList<>(); String put = "put sys.cpu.user " + TEST_TIME + " 1.0 tag1=value1 viz=a&b tag2=value2"; ByteBuf buf = Unpooled.wrappedBuffer(put.getBytes()); decoder.decode(null, buf, results); Assert.assertEquals(1, results.size()); Assert.assertEquals(MetricRequest.class, results.get(0).getClass()); Metric m = ((MetricRequest) results.get(0)).getMetric(); Metric expected = Metric.newBuilder() .name("sys.cpu.user") .value(TEST_TIME, 1.0D) .tag(new Tag("tag1", "value1")).tag(new Tag("tag2", "value2")) .tag(MetricAdapter.VISIBILITY_TAG, "a&b") .build(); Assert.assertTrue(expected.equals(m)); }
@Test public void testVersion() throws Exception { TcpDecoder decoder = new TcpDecoder(); List<Object> results = new ArrayList<>(); String put = "version"; ByteBuf buf = Unpooled.wrappedBuffer(put.getBytes()); decoder.decode(null, buf, results); Assert.assertEquals(1, results.size()); Assert.assertEquals(VersionRequest.class, results.get(0).getClass()); }
@Test public void testUnknownOperation() throws Exception { TcpDecoder decoder = new TcpDecoder(); List<Object> results = new ArrayList<>(); String put = "foo"; ByteBuf buf = Unpooled.wrappedBuffer(put.getBytes()); decoder.decode(null, buf, results); Assert.assertEquals(0, results.size()); }