iron-spi-aws/src/main/java/io/axway/iron/spi/aws/kinesis/AwsKinesisUtils.java
private static void waitStreamActivation(AmazonKinesis consumer, String streamName, long streamCreationTimeoutMillis) {
        DescribeStreamRequest describeStreamRequest = new DescribeStreamRequest().withStreamName(streamName).withLimit(1);
        DescribeStreamResult describeStreamResult = null;
        String streamStatus = null;
        long endTime = System.currentTimeMillis() + streamCreationTimeoutMillis;
        do {
            try {
                describeStreamResult = consumer.describeStream(describeStreamRequest);
                streamStatus = describeStreamResult.getStreamDescription().getStreamStatus();
                if (ACTIVE_STREAM_STATUS.equals(streamStatus)) {
                    break;
                }
                Thread.sleep(100);
            } catch (ResourceNotFoundException | LimitExceededException ignored) {
                // ignored
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
                throw new AwsKinesisException("Thread interrupted while waiting for stream activation", args -> args.add("streamName", streamName), e);
            }
        } while (System.currentTimeMillis() < endTime);
        if (describeStreamResult == null || streamStatus == null || !streamStatus.equals(ACTIVE_STREAM_STATUS)) {
            throw new AwsKinesisException("Stream never went active",
                                          args -> args.add("streamName", streamName).add("streamCreationTimeoutMillis", streamCreationTimeoutMillis));
        }
    }
private static void waitStreamActivation(AmazonKinesis consumer, String streamName, long streamCreationTimeoutMillis) {
        DescribeStreamRequest describeStreamRequest = new DescribeStreamRequest().withStreamName(streamName).withLimit(1);
        DescribeStreamResult describeStreamResult = null;
        String streamStatus = null;
        long endTime = System.currentTimeMillis() + streamCreationTimeoutMillis;
        do {
            try {
                describeStreamResult = consumer.describeStream(describeStreamRequest);
                streamStatus = describeStreamResult.getStreamDescription().getStreamStatus();
                if (ACTIVE_STREAM_STATUS.equals(streamStatus)) {
                    break;
                }
                Thread.sleep(100);
            } catch (ResourceNotFoundException | LimitExceededException ignored) {
                // ignored
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
                throw new AwsKinesisException("Thread interrupted while waiting for stream activation", args -> args.add("streamName", streamName), e);
            }
        } while (System.currentTimeMillis() < endTime);
        if (describeStreamResult == null || streamStatus == null || !streamStatus.equals(ACTIVE_STREAM_STATUS)) {
            throw new AwsKinesisException("Stream never went active",
                                          args -> args.add("streamName", streamName).add("streamCreationTimeoutMillis", streamCreationTimeoutMillis));
        }
    }
private static void waitStreamActivation ( AmazonKinesis consumer , String streamName , long streamCreationTimeoutMillis ) { DescribeStreamRequest describeStreamRequest = new DescribeStreamRequest ( ) . withStreamName ( streamName ) . withLimit ( 1 ) ; DescribeStreamResult describeStreamResult = null ; String streamStatus = null ; long endTime = System . currentTimeMillis ( ) + streamCreationTimeoutMillis ; do { try { describeStreamResult = consumer . describeStream ( describeStreamRequest ) ; streamStatus = describeStreamResult . getStreamDescription ( ) . getStreamStatus ( ) ; if ( ACTIVE_STREAM_STATUS . equals ( streamStatus ) ) { break ; } Thread . sleep ( 100 ) ; } catch ( ResourceNotFoundException | LimitExceededException ignored ) { // ignored } catch ( InterruptedException e ) { Thread . currentThread ( ) . interrupt ( ) ; throw new AwsKinesisException ( "Thread interrupted while waiting for stream activation" , args -> args . add ( "streamName" , streamName ) , e ) ; } } while ( System . currentTimeMillis ( ) < endTime ) ; if ( describeStreamResult == null || streamStatus == null || ! streamStatus . equals ( ACTIVE_STREAM_STATUS ) ) { throw new AwsKinesisException ( "Stream never went active" , args -> args . add ( "streamName" , streamName ) . add ( "streamCreationTimeoutMillis" , streamCreationTimeoutMillis ) ) ; } }
Waits that the stream has been created.
Waits that the stream has been created .