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 shouldParseHeaderProto() {
    // When:
    final List<StreamedRow> rows = KsqlTargetUtil.toRows(Buffer.buffer("[{\"header\":{\"queryId\":\"queryId\"," +
            "\"schema\":\"`A` INTEGER KEY, `B` DOUBLE, `C` ARRAY<STRING>\"," +
            "\"protoSchema\":" +
            "\"syntax = \\\"proto3\\\";\\n" +
            "\\n" +
            "message ConnectDefault1 {\\n" +
            "  int32 A = 1;\\n" +
            "  double B = 2;\\n" +
            "  repeated string C = 3;\\n" +
            "}\\n" +
            "\"}}]"), Functions.identity());

    StreamedRow row = rows.get(0);

    // Then:
    assertThat(row.getHeader().isPresent(), is(true));
    assertThat(row.getHeader().get().getQueryId().toString(), is("queryId"));

    assertThat(row.getHeader().get().getSchema().toString(), is("`A` INTEGER KEY, `B` DOUBLE, `C` ARRAY<STRING>"));
    assertThat(row.getHeader().get().getProtoSchema().get(), is("syntax = \"proto3\";\n\nmessage ConnectDefault1 {\n  int32 A = 1;\n  double B = 2;\n  repeated string C = 3;\n}\n"));
  }