public static List<StreamedRow> toRows(
      final Buffer buff,
      final Function<StreamedRow, StreamedRow> addHostInfo
  ) {

    final List<StreamedRow> rows = new ArrayList<>();
    int begin = 0;

    for (int i = 0; i <= buff.length(); i++) {
      if ((i == buff.length() && (i - begin > 1))
          || (i < buff.length() && buff.getByte(i) == (byte) '\n')) {
        if (begin != i) { // Ignore random newlines - the server can send these
          final Buffer sliced = buff.slice(begin, i);
          final Buffer tidied = toJsonMsg(sliced, true);
          if (tidied.length() > 0) {
            final StreamedRow row = deserialize(tidied, StreamedRow.class);
            rows.add(addHostInfo.apply(row));
          }
        }

        begin = i + 1;
      }
    }
    return rows;
  }
@Test
  public void toRows() {
    // When:
    final List<StreamedRow> rows =  KsqlTargetUtil.toRows(Buffer.buffer(
        "[{\"header\":{\"queryId\":\"query_id_10\",\"schema\":\"`col1` STRING\"}},\n"
        + "{\"row\":{\"columns\":[\"Row1\"]}},\n"
        + "{\"row\":{\"columns\":[\"Row2\"]}},\n"), Functions.identity());

    // Then:
    assertThat(rows.size(), is(3));
    final StreamedRow row = rows.get(0);
    assertThat(row.getHeader().isPresent(), is(true));
    assertThat(row.getHeader().get().getQueryId().toString(), is("query_id_10"));
    assertThat(row.getHeader().get().getSchema().key(), is(Collections.emptyList()));
    assertThat(row.getHeader().get().getSchema().value().size(), is(1));
    assertThat(row.getHeader().get().getSchema().value().get(0),
        is (Column.of(ColumnName.of("col1"), SqlTypes.STRING, Namespace.VALUE, 0)));
    final StreamedRow row2 = rows.get(1);
    assertThat(row2.getRow().isPresent(), is(true));
    assertThat(row2.getRow().get().getColumns(), is(ImmutableList.of("Row1")));
    final StreamedRow row3 = rows.get(2);
    assertThat(row3.getRow().isPresent(), is(true));
    assertThat(row3.getRow().get().getColumns(), is(ImmutableList.of("Row2")));

  }