protected function getMessages($shard_id = NULL, $last_sequence_number = NULL): array
    {

        $records = [];

        // Get the initial iterator.
        try {
            $shard_iterator = $this->getInitialShardIterator($shard_id, $last_sequence_number);
        } catch (\Exception $e) {
            $this->processError($e, [
                'shard_id' => $shard_id,
                'last_sequence_number' => $last_sequence_number,
            ]);

            return $records;
        }

        do {

            $has_records = FALSE;

            // Load batch of records.
            try {
                $res = $this->getClient()->getRecords([
                    'Limit' => $this->batchSize,
                    'ShardIterator' => $shard_iterator,
                    'StreamName' => $this->getStreamName(),
                ]);

                // Get the Shard iterator for next batch.
                $shard_iterator = $res->get('NextShardIterator');
                $behind_latest = $res->get('MillisBehindLatest');

                foreach ($res->search('Records[].[SequenceNumber, Data]') as $event) {
                    list($sequence_number, $message_data) = $event;

                    try {
                        $records[$sequence_number] = $this->getMessageFactory()->unserialize($message_data);
                    } catch (\Exception $e) {
                        $this->processError($e, $event);
                    }

                    $has_records = TRUE;
                }
            } catch (\Exception $e) {
                $this->processError($e);
            }

        } while (!empty($shard_iterator) && !empty($behind_latest) && !$has_records);

        if (!empty($sequence_number)) {
            $this->lastSequenceNumber = $sequence_number;
        }

        if (!empty($behind_latest)) {
            $this->lag = $behind_latest;
        }

        return $records;
    }
protected function getMessages($shard_id = NULL, $last_sequence_number = NULL): array
    {

        $records = [];

        // Get the initial iterator.
        try {
            $shard_iterator = $this->getInitialShardIterator($shard_id, $last_sequence_number);
        } catch (\Exception $e) {
            $this->processError($e, [
                'shard_id' => $shard_id,
                'last_sequence_number' => $last_sequence_number,
            ]);

            return $records;
        }

        do {

            $has_records = FALSE;

            // Load batch of records.
            try {
                $res = $this->getClient()->getRecords([
                    'Limit' => $this->batchSize,
                    'ShardIterator' => $shard_iterator,
                    'StreamName' => $this->getStreamName(),
                ]);

                // Get the Shard iterator for next batch.
                $shard_iterator = $res->get('NextShardIterator');
                $behind_latest = $res->get('MillisBehindLatest');

                foreach ($res->search('Records[].[SequenceNumber, Data]') as $event) {
                    list($sequence_number, $message_data) = $event;

                    try {
                        $records[$sequence_number] = $this->getMessageFactory()->unserialize($message_data);
                    } catch (\Exception $e) {
                        $this->processError($e, $event);
                    }

                    $has_records = TRUE;
                }
            } catch (\Exception $e) {
                $this->processError($e);
            }

        } while (!empty($shard_iterator) && !empty($behind_latest) && !$has_records);

        if (!empty($sequence_number)) {
            $this->lastSequenceNumber = $sequence_number;
        }

        if (!empty($behind_latest)) {
            $this->lag = $behind_latest;
        }

        return $records;
    }
protected function getMessages ( $ shard_id = NULL , $ last_sequence_number = NULL ) : array { $ records = [ ] ; // Get the initial iterator. try { $ shard_iterator = $ this -> getInitialShardIterator ( $ shard_id , $ last_sequence_number ) ; } catch ( \ Exception $ e ) { $ this -> processError ( $ e , [ 'shard_id' => $ shard_id , 'last_sequence_number' => $ last_sequence_number , ] ) ; return $ records ; } do { $ has_records = FALSE ; // Load batch of records. try { $ res = $ this -> getClient ( ) -> getRecords ( [ 'Limit' => $ this -> batchSize , 'ShardIterator' => $ shard_iterator , 'StreamName' => $ this -> getStreamName ( ) , ] ) ; // Get the Shard iterator for next batch. $ shard_iterator = $ res -> get ( 'NextShardIterator' ) ; $ behind_latest = $ res -> get ( 'MillisBehindLatest' ) ; foreach ( $ res -> search ( 'Records[].[SequenceNumber, Data]' ) as $ event ) { list ( $ sequence_number , $ message_data ) = $ event ; try { $ records [ $ sequence_number ] = $ this -> getMessageFactory ( ) -> unserialize ( $ message_data ) ; } catch ( \ Exception $ e ) { $ this -> processError ( $ e , $ event ) ; } $ has_records = TRUE ; } } catch ( \ Exception $ e ) { $ this -> processError ( $ e ) ; } } while ( ! empty ( $ shard_iterator ) && ! empty ( $ behind_latest ) && ! $ has_records ) ; if ( ! empty ( $ sequence_number ) ) { $ this -> lastSequenceNumber = $ sequence_number ; } if ( ! empty ( $ behind_latest ) ) { $ this -> lag = $ behind_latest ; } return $ records ; }
Do a single call to a Kinesis stream to get messages.

@param string $shard_id
The shard id from which to start reading.
@param string $last_sequence_number
The last record sequence number the last call of getRecords() ended. The
value will be provided by a subsequent call of getLastSequenceNumber().

@return \robocloud\Message\MessageInterface[]
The messages.
Do a single call to a Kinesis stream to get messages .