@Test void responseWithMultipleValues() throws IOException { String data = "#datatype,string,long,dateTime:RFC3339,dateTime:RFC3339,string,string,string,string,long,long,string\n" + "#group,false,false,true,true,true,true,true,true,false,false,false\n" + "#default,_result,,,,,,,,,,\n" + ",result,table,_start,_stop,_field,_measurement,host,region,_value2,value1,value_str\n" + ",,0,1677-09-21T00:12:43.145224192Z,2018-07-16T11:21:02.547596934Z,free,mem,A,west,121,11,test\n" + ",,1,1677-09-21T00:12:43.145224192Z,2018-07-16T11:21:02.547596934Z,free,mem,B,west,484,22,test\n" + ",,2,1677-09-21T00:12:43.145224192Z,2018-07-16T11:21:02.547596934Z,usage_system,cpu,A,west,1444,38,test\n" + ",,3,1677-09-21T00:12:43.145224192Z,2018-07-16T11:21:02.547596934Z,user_usage,cpu,A,west,2401,49,test"; List<FluxTable> tables = parseFluxResponse(data); List<FluxColumn> columnHeaders = tables.get(0).getColumns(); Assertions.assertThat(columnHeaders).hasSize(11); FluxColumn fluxColumn1 = columnHeaders.get(0); LOG.info("FluxColumn1: " + fluxColumn1); Assertions.assertThat(fluxColumn1.isGroup()).isFalse(); Assertions.assertThat(columnHeaders.get(1).isGroup()).isFalse(); Assertions.assertThat(columnHeaders.get(2).isGroup()).isTrue(); Assertions.assertThat(columnHeaders.get(3).isGroup()).isTrue(); Assertions.assertThat(columnHeaders.get(4).isGroup()).isTrue(); Assertions.assertThat(columnHeaders.get(5).isGroup()).isTrue(); Assertions.assertThat(columnHeaders.get(6).isGroup()).isTrue(); Assertions.assertThat(columnHeaders.get(7).isGroup()).isTrue(); Assertions.assertThat(columnHeaders.get(8).isGroup()).isFalse(); Assertions.assertThat(columnHeaders.get(9).isGroup()).isFalse(); Assertions.assertThat(columnHeaders.get(10).isGroup()).isFalse(); Assertions.assertThat(tables).hasSize(4); FluxTable fluxTable1 = tables.get(0); LOG.info("FluxTable1: " + fluxTable1); Assertions.assertThat(fluxTable1.getRecords()).hasSize(1); FluxRecord fluxRecord1 = fluxTable1.getRecords().get(0); LOG.info("FluxRecord1: " + fluxRecord1); Assertions.assertThat(0).isEqualTo(fluxRecord1.getTable()); Assertions.assertThat(fluxRecord1.getValues()) .hasEntrySatisfying("host", value -> Assertions.assertThat(value).isEqualTo("A")) .hasEntrySatisfying("region", value -> Assertions.assertThat(value).isEqualTo("west")); Assertions.assertThat(fluxRecord1.getValues()).hasSize(11); Assertions.assertThat(fluxRecord1.getValue()).isNull(); Assertions.assertThat(fluxRecord1.getValues()) .hasEntrySatisfying("value1", value -> Assertions.assertThat(value).isEqualTo(11L)) .hasEntrySatisfying("_value2", value -> Assertions.assertThat(value).isEqualTo(121L)) .hasEntrySatisfying("value_str", value -> Assertions.assertThat(value).isEqualTo("test")); Assertions.assertThat(fluxRecord1.getValueByIndex(8)).isEqualTo(121L); Assertions.assertThat(fluxRecord1.getValueByIndex(9)).isEqualTo(11L); Assertions.assertThat(fluxRecord1.getValueByIndex(10)).isEqualTo("test"); FluxTable fluxTable2 = tables.get(1); LOG.info("FluxTable2: " + fluxTable2); Assertions.assertThat(fluxTable2.getRecords()).hasSize(1); FluxRecord fluxRecord2 = fluxTable2.getRecords().get(0); Assertions.assertThat(1).isEqualTo(fluxRecord2.getTable()); Assertions.assertThat(fluxRecord2.getValues()) .hasEntrySatisfying("host", value -> Assertions.assertThat(value).isEqualTo("B")) .hasEntrySatisfying("region", value -> Assertions.assertThat(value).isEqualTo("west")); Assertions.assertThat(fluxRecord2.getValues()).hasSize(11); Assertions.assertThat(fluxRecord2.getValue()).isNull(); Assertions.assertThat(fluxRecord2.getValues()) .hasEntrySatisfying("value1", value -> Assertions.assertThat(value).isEqualTo(22L)) .hasEntrySatisfying("_value2", value -> Assertions.assertThat(value).isEqualTo(484L)) .hasEntrySatisfying("value_str", value -> Assertions.assertThat(value).isEqualTo("test")); FluxTable fluxTable3 = tables.get(2); LOG.info("FluxTable3: " + fluxTable3); Assertions.assertThat(fluxTable3.getRecords()).hasSize(1); FluxRecord fluxRecord3 = fluxTable3.getRecords().get(0); Assertions.assertThat(2).isEqualTo(fluxRecord3.getTable()); Assertions.assertThat(fluxRecord3.getValues()) .hasEntrySatisfying("host", value -> Assertions.assertThat(value).isEqualTo("A")) .hasEntrySatisfying("region", value -> Assertions.assertThat(value).isEqualTo("west")); Assertions.assertThat(fluxRecord3.getValues()).hasSize(11); Assertions.assertThat(fluxRecord3.getValue()).isNull(); Assertions.assertThat(fluxRecord3.getValues()) .hasEntrySatisfying("value1", value -> Assertions.assertThat(value).isEqualTo(38L)) .hasEntrySatisfying("_value2", value -> Assertions.assertThat(value).isEqualTo(1444L)) .hasEntrySatisfying("value_str", value -> Assertions.assertThat(value).isEqualTo("test")); FluxTable fluxTable4 = tables.get(3); LOG.info("FluxTable4: " + fluxTable4); Assertions.assertThat(fluxTable4.getRecords()).hasSize(1); FluxRecord fluxRecord4 = fluxTable4.getRecords().get(0); Assertions.assertThat(3).isEqualTo(fluxRecord4.getTable()); Assertions.assertThat(fluxRecord4.getValues()) .hasEntrySatisfying("host", value -> Assertions.assertThat(value).isEqualTo("A")) .hasEntrySatisfying("region", value -> Assertions.assertThat(value).isEqualTo("west")); Assertions.assertThat(fluxRecord4.getValues()).hasSize(11); Assertions.assertThat(fluxRecord4.getValue()).isNull(); Assertions.assertThat(fluxRecord4.getValues()) .hasEntrySatisfying("value1", value -> Assertions.assertThat(value).isEqualTo(49L)) .hasEntrySatisfying("_value2", value -> Assertions.assertThat(value).isEqualTo(2401L)) .hasEntrySatisfying("value_str", value -> Assertions.assertThat(value).isEqualTo("test")); }
public void parseFluxResponse(@Nonnull final BufferedSource bufferedSource, @Nonnull final Cancellable cancellable, @Nonnull final FluxResponseConsumer consumer) throws IOException { Arguments.checkNotNull(bufferedSource, "bufferedSource"); Reader reader = new InputStreamReader(bufferedSource.inputStream()); ParsingState parsingState = ParsingState.NORMAL; try (CSVParser parser = new CSVParser(reader, CSVFormat.DEFAULT)) { int tableIndex = 0; int tableId = -1; boolean startNewTable = false; FluxTable table = null; for (CSVRecord csvRecord : parser) { if (cancellable.isCancelled()) { return; } long recordNumber = csvRecord.getRecordNumber(); if (ERROR_RECORD_INDEX == recordNumber && csvRecord.get(1).equals("error") && csvRecord.get(2).equals("reference")) { parsingState = ParsingState.IN_ERROR; continue; } if (ParsingState.IN_ERROR.equals(parsingState)) { String error = csvRecord.get(1); String referenceValue = csvRecord.get(2); int reference = 0; if (referenceValue != null && !referenceValue.isEmpty()) { reference = Integer.valueOf(referenceValue); } throw new FluxQueryException(error, reference); } String token = csvRecord.get(0); if ("#datatype".equals(token)) { startNewTable = true; table = new FluxTable(); consumer.accept(tableIndex, cancellable, table); tableIndex++; tableId = -1; } else if (table == null) { String message = "Unable to parse CSV response. FluxTable definition was not found."; throw new FluxCsvParserException(message); } if ("#datatype".equals(token)) { addDataTypes(table, toList(csvRecord)); } else if ("#group".equals(token)) { addGroups(table, toList(csvRecord)); } else if ("#default".equals(token)) { addDefaultEmptyValues(table, toList(csvRecord)); } else { if (startNewTable) { addColumnNamesAndTags(table, toList(csvRecord)); startNewTable = false; continue; } int currentId = Integer.parseInt(csvRecord.get(1 + 1)); if (tableId == -1) { tableId = currentId; } if (tableId != currentId) { List<FluxColumn> fluxColumns = table.getColumns(); table = new FluxTable(); table.getColumns().addAll(fluxColumns); consumer.accept(tableIndex, cancellable, table); tableIndex++; tableId = currentId; } FluxRecord fluxRecord = parseRecord(tableIndex - 1, table, csvRecord); consumer.accept(tableIndex - 1, cancellable, fluxRecord); } } } }
FluxCsvParser { public void parseFluxResponse(@Nonnull final BufferedSource bufferedSource, @Nonnull final Cancellable cancellable, @Nonnull final FluxResponseConsumer consumer) throws IOException { Arguments.checkNotNull(bufferedSource, "bufferedSource"); Reader reader = new InputStreamReader(bufferedSource.inputStream()); ParsingState parsingState = ParsingState.NORMAL; try (CSVParser parser = new CSVParser(reader, CSVFormat.DEFAULT)) { int tableIndex = 0; int tableId = -1; boolean startNewTable = false; FluxTable table = null; for (CSVRecord csvRecord : parser) { if (cancellable.isCancelled()) { return; } long recordNumber = csvRecord.getRecordNumber(); if (ERROR_RECORD_INDEX == recordNumber && csvRecord.get(1).equals("error") && csvRecord.get(2).equals("reference")) { parsingState = ParsingState.IN_ERROR; continue; } if (ParsingState.IN_ERROR.equals(parsingState)) { String error = csvRecord.get(1); String referenceValue = csvRecord.get(2); int reference = 0; if (referenceValue != null && !referenceValue.isEmpty()) { reference = Integer.valueOf(referenceValue); } throw new FluxQueryException(error, reference); } String token = csvRecord.get(0); if ("#datatype".equals(token)) { startNewTable = true; table = new FluxTable(); consumer.accept(tableIndex, cancellable, table); tableIndex++; tableId = -1; } else if (table == null) { String message = "Unable to parse CSV response. FluxTable definition was not found."; throw new FluxCsvParserException(message); } if ("#datatype".equals(token)) { addDataTypes(table, toList(csvRecord)); } else if ("#group".equals(token)) { addGroups(table, toList(csvRecord)); } else if ("#default".equals(token)) { addDefaultEmptyValues(table, toList(csvRecord)); } else { if (startNewTable) { addColumnNamesAndTags(table, toList(csvRecord)); startNewTable = false; continue; } int currentId = Integer.parseInt(csvRecord.get(1 + 1)); if (tableId == -1) { tableId = currentId; } if (tableId != currentId) { List<FluxColumn> fluxColumns = table.getColumns(); table = new FluxTable(); table.getColumns().addAll(fluxColumns); consumer.accept(tableIndex, cancellable, table); tableIndex++; tableId = currentId; } FluxRecord fluxRecord = parseRecord(tableIndex - 1, table, csvRecord); consumer.accept(tableIndex - 1, cancellable, fluxRecord); } } } } }
FluxCsvParser { public void parseFluxResponse(@Nonnull final BufferedSource bufferedSource, @Nonnull final Cancellable cancellable, @Nonnull final FluxResponseConsumer consumer) throws IOException { Arguments.checkNotNull(bufferedSource, "bufferedSource"); Reader reader = new InputStreamReader(bufferedSource.inputStream()); ParsingState parsingState = ParsingState.NORMAL; try (CSVParser parser = new CSVParser(reader, CSVFormat.DEFAULT)) { int tableIndex = 0; int tableId = -1; boolean startNewTable = false; FluxTable table = null; for (CSVRecord csvRecord : parser) { if (cancellable.isCancelled()) { return; } long recordNumber = csvRecord.getRecordNumber(); if (ERROR_RECORD_INDEX == recordNumber && csvRecord.get(1).equals("error") && csvRecord.get(2).equals("reference")) { parsingState = ParsingState.IN_ERROR; continue; } if (ParsingState.IN_ERROR.equals(parsingState)) { String error = csvRecord.get(1); String referenceValue = csvRecord.get(2); int reference = 0; if (referenceValue != null && !referenceValue.isEmpty()) { reference = Integer.valueOf(referenceValue); } throw new FluxQueryException(error, reference); } String token = csvRecord.get(0); if ("#datatype".equals(token)) { startNewTable = true; table = new FluxTable(); consumer.accept(tableIndex, cancellable, table); tableIndex++; tableId = -1; } else if (table == null) { String message = "Unable to parse CSV response. FluxTable definition was not found."; throw new FluxCsvParserException(message); } if ("#datatype".equals(token)) { addDataTypes(table, toList(csvRecord)); } else if ("#group".equals(token)) { addGroups(table, toList(csvRecord)); } else if ("#default".equals(token)) { addDefaultEmptyValues(table, toList(csvRecord)); } else { if (startNewTable) { addColumnNamesAndTags(table, toList(csvRecord)); startNewTable = false; continue; } int currentId = Integer.parseInt(csvRecord.get(1 + 1)); if (tableId == -1) { tableId = currentId; } if (tableId != currentId) { List<FluxColumn> fluxColumns = table.getColumns(); table = new FluxTable(); table.getColumns().addAll(fluxColumns); consumer.accept(tableIndex, cancellable, table); tableIndex++; tableId = currentId; } FluxRecord fluxRecord = parseRecord(tableIndex - 1, table, csvRecord); consumer.accept(tableIndex - 1, cancellable, fluxRecord); } } } }  }
FluxCsvParser { public void parseFluxResponse(@Nonnull final BufferedSource bufferedSource, @Nonnull final Cancellable cancellable, @Nonnull final FluxResponseConsumer consumer) throws IOException { Arguments.checkNotNull(bufferedSource, "bufferedSource"); Reader reader = new InputStreamReader(bufferedSource.inputStream()); ParsingState parsingState = ParsingState.NORMAL; try (CSVParser parser = new CSVParser(reader, CSVFormat.DEFAULT)) { int tableIndex = 0; int tableId = -1; boolean startNewTable = false; FluxTable table = null; for (CSVRecord csvRecord : parser) { if (cancellable.isCancelled()) { return; } long recordNumber = csvRecord.getRecordNumber(); if (ERROR_RECORD_INDEX == recordNumber && csvRecord.get(1).equals("error") && csvRecord.get(2).equals("reference")) { parsingState = ParsingState.IN_ERROR; continue; } if (ParsingState.IN_ERROR.equals(parsingState)) { String error = csvRecord.get(1); String referenceValue = csvRecord.get(2); int reference = 0; if (referenceValue != null && !referenceValue.isEmpty()) { reference = Integer.valueOf(referenceValue); } throw new FluxQueryException(error, reference); } String token = csvRecord.get(0); if ("#datatype".equals(token)) { startNewTable = true; table = new FluxTable(); consumer.accept(tableIndex, cancellable, table); tableIndex++; tableId = -1; } else if (table == null) { String message = "Unable to parse CSV response. FluxTable definition was not found."; throw new FluxCsvParserException(message); } if ("#datatype".equals(token)) { addDataTypes(table, toList(csvRecord)); } else if ("#group".equals(token)) { addGroups(table, toList(csvRecord)); } else if ("#default".equals(token)) { addDefaultEmptyValues(table, toList(csvRecord)); } else { if (startNewTable) { addColumnNamesAndTags(table, toList(csvRecord)); startNewTable = false; continue; } int currentId = Integer.parseInt(csvRecord.get(1 + 1)); if (tableId == -1) { tableId = currentId; } if (tableId != currentId) { List<FluxColumn> fluxColumns = table.getColumns(); table = new FluxTable(); table.getColumns().addAll(fluxColumns); consumer.accept(tableIndex, cancellable, table); tableIndex++; tableId = currentId; } FluxRecord fluxRecord = parseRecord(tableIndex - 1, table, csvRecord); consumer.accept(tableIndex - 1, cancellable, fluxRecord); } } } }  void parseFluxResponse(@Nonnull final BufferedSource bufferedSource,
                                  @Nonnull final Cancellable cancellable,
                                  @Nonnull final FluxResponseConsumer consumer); }
FluxCsvParser { public void parseFluxResponse(@Nonnull final BufferedSource bufferedSource, @Nonnull final Cancellable cancellable, @Nonnull final FluxResponseConsumer consumer) throws IOException { Arguments.checkNotNull(bufferedSource, "bufferedSource"); Reader reader = new InputStreamReader(bufferedSource.inputStream()); ParsingState parsingState = ParsingState.NORMAL; try (CSVParser parser = new CSVParser(reader, CSVFormat.DEFAULT)) { int tableIndex = 0; int tableId = -1; boolean startNewTable = false; FluxTable table = null; for (CSVRecord csvRecord : parser) { if (cancellable.isCancelled()) { return; } long recordNumber = csvRecord.getRecordNumber(); if (ERROR_RECORD_INDEX == recordNumber && csvRecord.get(1).equals("error") && csvRecord.get(2).equals("reference")) { parsingState = ParsingState.IN_ERROR; continue; } if (ParsingState.IN_ERROR.equals(parsingState)) { String error = csvRecord.get(1); String referenceValue = csvRecord.get(2); int reference = 0; if (referenceValue != null && !referenceValue.isEmpty()) { reference = Integer.valueOf(referenceValue); } throw new FluxQueryException(error, reference); } String token = csvRecord.get(0); if ("#datatype".equals(token)) { startNewTable = true; table = new FluxTable(); consumer.accept(tableIndex, cancellable, table); tableIndex++; tableId = -1; } else if (table == null) { String message = "Unable to parse CSV response. FluxTable definition was not found."; throw new FluxCsvParserException(message); } if ("#datatype".equals(token)) { addDataTypes(table, toList(csvRecord)); } else if ("#group".equals(token)) { addGroups(table, toList(csvRecord)); } else if ("#default".equals(token)) { addDefaultEmptyValues(table, toList(csvRecord)); } else { if (startNewTable) { addColumnNamesAndTags(table, toList(csvRecord)); startNewTable = false; continue; } int currentId = Integer.parseInt(csvRecord.get(1 + 1)); if (tableId == -1) { tableId = currentId; } if (tableId != currentId) { List<FluxColumn> fluxColumns = table.getColumns(); table = new FluxTable(); table.getColumns().addAll(fluxColumns); consumer.accept(tableIndex, cancellable, table); tableIndex++; tableId = currentId; } FluxRecord fluxRecord = parseRecord(tableIndex - 1, table, csvRecord); consumer.accept(tableIndex - 1, cancellable, fluxRecord); } } } }  void parseFluxResponse(@Nonnull final BufferedSource bufferedSource,
                                  @Nonnull final Cancellable cancellable,
                                  @Nonnull final FluxResponseConsumer consumer);  }