JsonRecordReader implements RecordReader<String> { @Override public JsonRecord readRecord() { if (hasNextRecord()) { StringWriter stringWriter = new StringWriter(); JsonGenerator jsonGenerator = jsonGeneratorFactory.createGenerator(stringWriter); writeRecordStart(jsonGenerator); do { moveToNextElement(jsonGenerator); } while (!isEndRootObject()); if (arrayDepth != 2) { jsonGenerator.writeEnd(); } jsonGenerator.close(); Header header = new Header(++currentRecordNumber, getDataSourceName(), LocalDateTime.now()); return new JsonRecord(header, stringWriter.toString()); } else { return null; } } JsonRecordReader(final InputStream inputStream);  JsonRecordReader(final InputStream inputStream, final Charset charset); @Override void open(); @Override JsonRecord readRecord(); @Override void close();  }
@Test public void parsedJsonRecordPayloadShouldBeTheSameAsInTheDataSource() throws Exception { String expectedJson = "{\"id\":1,\"user\":\"foo\",\"message\":\"Hello\"}"; JsonRecord jsonRecord = jsonRecordReader.readRecord(); assertThat(jsonRecord).isNotNull(); assertThat(jsonRecord.getHeader().getNumber()).isEqualTo(1); assertThat(jsonRecord.getPayload()).isEqualTo(expectedJson); expectedJson = "{\"id\":2,\"user\":\"bar\",\"message\":\"Hi!\"}"; jsonRecord = jsonRecordReader.readRecord(); assertThat(jsonRecord).isNotNull(); assertThat(jsonRecord.getHeader().getNumber()).isEqualTo(2); assertThat(jsonRecord.getPayload()).isEqualTo(expectedJson); expectedJson = "{\"id\":3,\"user\":\"toto\",\"message\":\"yep ;-)\"}"; jsonRecord = jsonRecordReader.readRecord(); assertThat(jsonRecord).isNotNull(); assertThat(jsonRecord.getHeader().getNumber()).isEqualTo(3); assertThat(jsonRecord.getPayload()).isEqualTo(expectedJson); assertThat(jsonRecordReader.readRecord()).isNull(); }