InfluxDbWriteObjectSerializer { public String getGroupedLineProtocolString(InfluxDbWriteObject influxDbWriteObject, String realMeasurement) { HashSet<Long> times = new HashSet<>(); for (InfluxDbPoint point : influxDbWriteObject.getPoints()) { times.add(point.getTime()); } StringBuilder stringBuilder = new StringBuilder(); for (Long time : times) { Map<String, Object> fields = new HashMap<>(); Map<String, String> tags = null; for (InfluxDbPoint point : influxDbWriteObject.getPoints()) { if (point.getTime().equals(time)) { mergeFields(fields, point.getFields(), point.getMeasurement()); tags = point.getTags(); } } lineProtocol(tags, fields, realMeasurement, time, influxDbWriteObject.getPrecision(), stringBuilder); stringBuilder.append("\n"); } return stringBuilder.toString(); } InfluxDbWriteObjectSerializer(String measurementPrefix); String getLineProtocolString(InfluxDbWriteObject influxDbWriteObject); String getGroupedLineProtocolString(InfluxDbWriteObject influxDbWriteObject, String realMeasurement);  }
@Test public void shouldSerializeUsingGroupedLineProtocol() { Map<String, String> tags = new HashMap<String, String>(); tags.put("tag1Key", "tag1Value"); Map<String, Object> fields = new HashMap<String, Object>(); fields.put("field1Key", "field1Value"); InfluxDbPoint point1 = new InfluxDbPoint("aaa.bbb.ccc", tags, 456l, fields); Set<InfluxDbPoint> set = new HashSet<InfluxDbPoint>(); set.add(point1); InfluxDbWriteObject influxDbWriteObject = mock(InfluxDbWriteObject.class); when(influxDbWriteObject.getPoints()).thenReturn(set); when(influxDbWriteObject.getPrecision()).thenReturn(TimeUnit.MICROSECONDS); InfluxDbWriteObjectSerializer influxDbWriteObjectSerializer = new InfluxDbWriteObjectSerializer(""); String lineString = influxDbWriteObjectSerializer.getGroupedLineProtocolString(influxDbWriteObject, "xxx"); assertThat(lineString).isEqualTo( "xxx,tag1Key=tag1Value bbb.ccc.field1Key=\"field1Value\" 456000\n"); InfluxDbPoint point2 = new InfluxDbPoint("www.yyy.zzz", tags, 456l, fields); set.add(point2); lineString = influxDbWriteObjectSerializer.getGroupedLineProtocolString(influxDbWriteObject, "xxx"); assertThat(lineString).isEqualTo( "xxx,tag1Key=tag1Value bbb.ccc.field1Key=\"field1Value\",yyy.zzz.field1Key=\"field1Value\" 456000\n"); }
@Test public void groupedLinesShouldFoldValueFields() { Map<String, String> tags = new HashMap<String, String>(); tags.put("tag1Key", "tag1Value"); Map<String, Object> fields1 = new HashMap<String, Object>(); fields1.put("value", "111"); Map<String, Object> fields2 = new HashMap<String, Object>(); fields2.put("value", "222"); InfluxDbPoint point1 = new InfluxDbPoint("aa.bb.cc1", tags, 456l, fields1); InfluxDbPoint point2 = new InfluxDbPoint("aa.bb.cc2", tags, 456l, fields2); Set<InfluxDbPoint> set = new HashSet<InfluxDbPoint>(); set.add(point1); set.add(point2); InfluxDbWriteObject influxDbWriteObject = mock(InfluxDbWriteObject.class); when(influxDbWriteObject.getPoints()).thenReturn(set); when(influxDbWriteObject.getPrecision()).thenReturn(TimeUnit.MICROSECONDS); InfluxDbWriteObjectSerializer influxDbWriteObjectSerializer = new InfluxDbWriteObjectSerializer(""); String lineString = influxDbWriteObjectSerializer.getGroupedLineProtocolString(influxDbWriteObject, "xxx"); assertThat(lineString).isEqualTo("xxx,tag1Key=tag1Value bb.cc2=\"222\",bb.cc1=\"111\" 456000\n"); }
@Test public void undottedMeasurementShouldFallback() { Map<String, String> tags = new HashMap<String, String>(); tags.put("tag1Key", "tag1Value"); Map<String, Object> fields = new HashMap<String, Object>(); fields.put("field1Key", "field1Value"); InfluxDbPoint point1 = new InfluxDbPoint("aaa", tags, 456l, fields); Set<InfluxDbPoint> set = new HashSet<InfluxDbPoint>(); set.add(point1); InfluxDbWriteObject influxDbWriteObject = mock(InfluxDbWriteObject.class); when(influxDbWriteObject.getPoints()).thenReturn(set); when(influxDbWriteObject.getPrecision()).thenReturn(TimeUnit.MICROSECONDS); InfluxDbWriteObjectSerializer influxDbWriteObjectSerializer = new InfluxDbWriteObjectSerializer(""); String lineString = influxDbWriteObjectSerializer.getGroupedLineProtocolString(influxDbWriteObject, "xxx"); assertThat(lineString).isEqualTo( "xxx,tag1Key=tag1Value aaa.field1Key=\"field1Value\" 456000\n"); InfluxDbPoint point2 = new InfluxDbPoint("bbb", tags, 456l, fields); set.add(point2); lineString = influxDbWriteObjectSerializer.getGroupedLineProtocolString(influxDbWriteObject, "xxx"); assertThat(lineString).isEqualTo( "xxx,tag1Key=tag1Value aaa.field1Key=\"field1Value\",bbb.field1Key=\"field1Value\" 456000\n"); }