SimplifiedKinesisClient { public String getShardIterator( final String streamName, final String shardId, final ShardIteratorType shardIteratorType, final String startingSequenceNumber, final Instant timestamp) throws TransientKinesisException { final Date date = timestamp != null ? timestamp.toDate() : null; return wrapExceptions( () -> kinesis .getShardIterator( new GetShardIteratorRequest() .withStreamName(streamName) .withShardId(shardId) .withShardIteratorType(shardIteratorType) .withStartingSequenceNumber(startingSequenceNumber) .withTimestamp(date)) .getShardIterator()); } SimplifiedKinesisClient(
      AmazonKinesis kinesis, AmazonCloudWatch cloudWatch, Integer limit); static SimplifiedKinesisClient from(AWSClientsProvider provider, Integer limit); String getShardIterator(
      final String streamName,
      final String shardId,
      final ShardIteratorType shardIteratorType,
      final String startingSequenceNumber,
      final Instant timestamp); List<Shard> listShards(final String streamName); GetKinesisRecordsResult getRecords(String shardIterator, String streamName, String shardId); GetKinesisRecordsResult getRecords(
      final String shardIterator,
      final String streamName,
      final String shardId,
      final Integer limit); long getBacklogBytes(String streamName, Instant countSince); long getBacklogBytes(
      final String streamName, final Instant countSince, final Instant countTo);  }
@Test public void shouldReturnIteratorStartingWithSequenceNumber() throws Exception { when(kinesis.getShardIterator( new GetShardIteratorRequest() .withStreamName(STREAM) .withShardId(SHARD_1) .withShardIteratorType(ShardIteratorType.AT_SEQUENCE_NUMBER) .withStartingSequenceNumber(SEQUENCE_NUMBER))) .thenReturn(new GetShardIteratorResult().withShardIterator(SHARD_ITERATOR)); String stream = underTest.getShardIterator( STREAM, SHARD_1, ShardIteratorType.AT_SEQUENCE_NUMBER, SEQUENCE_NUMBER, null); assertThat(stream).isEqualTo(SHARD_ITERATOR); }
@Test public void shouldReturnIteratorStartingWithTimestamp() throws Exception { Instant timestamp = Instant.now(); when(kinesis.getShardIterator( new GetShardIteratorRequest() .withStreamName(STREAM) .withShardId(SHARD_1) .withShardIteratorType(ShardIteratorType.AT_SEQUENCE_NUMBER) .withTimestamp(timestamp.toDate()))) .thenReturn(new GetShardIteratorResult().withShardIterator(SHARD_ITERATOR)); String stream = underTest.getShardIterator( STREAM, SHARD_1, ShardIteratorType.AT_SEQUENCE_NUMBER, null, timestamp); assertThat(stream).isEqualTo(SHARD_ITERATOR); }