@Test public void getRecords_GivenStreamWithSingleShard_ShouldReturnExpectedRecords() throws IOException { Shard shard = Shard.builder().shardId(UUID.randomUUID().toString()).build(); mockDescribeStream(shard); String shardIterator = mockGetShardIterator(shard.shardId()); List<SampleRecord> expected = Collections.singletonList(new SampleRecord()); mockGetRecords(shardIterator, null, expected); List<SampleRecord> actual = reader.getRecords(); Assert.assertEquals(expected, actual); }
@Override public List<T> getRecords() throws IOException { List<Shard> shards = this.client .describeStream(DescribeStreamRequest.builder().streamName(streamName).build()) .streamDescription().shards(); List<T> items = new LinkedList<>(); for (Shard shard : shards) { items.addAll(recordTranslator.toValues(getRecordsForShard(shard))); } return items; }
KinesisStreamReader implements Reader<T> { @Override public List<T> getRecords() throws IOException { List<Shard> shards = this.client .describeStream(DescribeStreamRequest.builder().streamName(streamName).build()) .streamDescription().shards(); List<T> items = new LinkedList<>(); for (Shard shard : shards) { items.addAll(recordTranslator.toValues(getRecordsForShard(shard))); } return items; } }
KinesisStreamReader implements Reader<T> { @Override public List<T> getRecords() throws IOException { List<Shard> shards = this.client .describeStream(DescribeStreamRequest.builder().streamName(streamName).build()) .streamDescription().shards(); List<T> items = new LinkedList<>(); for (Shard shard : shards) { items.addAll(recordTranslator.toValues(getRecordsForShard(shard))); } return items; } KinesisStreamReader(String streamName, AbstractKinesisRecordTranslator<T> recordTranslator);  KinesisStreamReader(KinesisClient client,
                               String streamName,
                               AbstractKinesisRecordTranslator<T> recordTranslator); }
KinesisStreamReader implements Reader<T> { @Override public List<T> getRecords() throws IOException { List<Shard> shards = this.client .describeStream(DescribeStreamRequest.builder().streamName(streamName).build()) .streamDescription().shards(); List<T> items = new LinkedList<>(); for (Shard shard : shards) { items.addAll(recordTranslator.toValues(getRecordsForShard(shard))); } return items; } KinesisStreamReader(String streamName, AbstractKinesisRecordTranslator<T> recordTranslator);  KinesisStreamReader(KinesisClient client,
                               String streamName,
                               AbstractKinesisRecordTranslator<T> recordTranslator); @Override List<T> getRecords(); @Override void setConfiguration(ReaderConfiguration configuration); }
KinesisStreamReader implements Reader<T> { @Override public List<T> getRecords() throws IOException { List<Shard> shards = this.client .describeStream(DescribeStreamRequest.builder().streamName(streamName).build()) .streamDescription().shards(); List<T> items = new LinkedList<>(); for (Shard shard : shards) { items.addAll(recordTranslator.toValues(getRecordsForShard(shard))); } return items; } KinesisStreamReader(String streamName, AbstractKinesisRecordTranslator<T> recordTranslator);  KinesisStreamReader(KinesisClient client,
                               String streamName,
                               AbstractKinesisRecordTranslator<T> recordTranslator); @Override List<T> getRecords(); @Override void setConfiguration(ReaderConfiguration configuration);  }