JpaRecordReader implements RecordReader<T> { @Override public GenericRecord<T> readRecord() { Header header = new Header(++currentRecordNumber, getDataSourceName(), LocalDateTime.now()); if (hasNextRecord()) { return new GenericRecord<>(header, iterator.next()); } else { return null; } } JpaRecordReader(final EntityManagerFactory entityManagerFactory, final String query, final Class<T> type); @Override void open(); @Override GenericRecord<T> readRecord(); @Override void close(); void setMaxResults(final int maxResults); static final int DEFAULT_MAX_RESULT; }
@Test public void testReadRecord() { Record<Tweet> record = jpaRecordReader.readRecord(); long recordNumber = record.getHeader().getNumber(); Tweet tweet = record.getPayload(); assertThat(recordNumber).isEqualTo(1); assertThat(tweet).isEqualTo(new Tweet(1, "foo", "easy batch rocks! #EasyBatch")); }
@Test public void testPaging() { int nbRecords = 0; while (jpaRecordReader.readRecord() != null) { nbRecords++; } assertThat(nbRecords).isEqualTo(2); }