OpentsdbMetricReader extends AbstractReader<RawMetric, Metric> { @Override public Stream<Metric> read(RawMetric rawMetric) { String bodyStr = readUTF8(rawMetric.getRawBody()); if (bodyStr.charAt(0) == ARRAY_START) { JSONArray array = JSON.parseArray(bodyStr); return array.stream() .map(o -> OpentsdbConverter.convertToModel((JSONObject) o)); } else { return Stream.of(OpentsdbConverter.convertToModel(JSON.parseObject(bodyStr))); } }  @Override Stream<Metric> read(RawMetric rawMetric);  }
@Test public void testRead() throws UnsupportedEncodingException { OpentsdbMetricReader reader = new OpentsdbMetricReader(); RawMetric metric = new RawMetric(); metric .setRawBody("{\"metric\":\"jvm.mem.free\",\"value\":1.2,\"timestamp\":1557844146565,\"tags\":{\"app\":\"demo\"}}" .getBytes("UTF-8")); Stream<Metric> metricStream = reader.read(metric); Metric m = metricStream.collect(Collectors.toList()).get(0); Assert.assertTrue(m.toString().contains("jvm.mem.free")); }