@Override
    public void run() {
        while (true) {
            int prefetchBatchSize;
            boolean nackQueueMessages = false;
            List<Message> messages = null;
            try {
                if (isClosed()) {
                    break;
                }
                
                synchronized (stateLock) {
                    waitForStart();
                    waitForPrefetch();
                    prefetchBatchSize = Math.min(numberOfMessagesToFetch(), SQSMessagingClientConstants.MAX_BATCH);
                }

                if (!isClosed()) {
                    messages = getMessages(prefetchBatchSize);
                }

                if (messages != null && !messages.isEmpty()) {
                    processReceivedMessages(messages);
                }
            } catch (InterruptedException e) {
                nackQueueMessages = true;
                break;
            } catch (Throwable e) {
                LOG.error("Unexpected exception when prefetch messages:", e);
                nackQueueMessages = true;
                throw new RuntimeException(e);
            } finally {
                if (isClosed() || nackQueueMessages) {
                    nackQueueMessages();
                }
            }
        }
    }
@ Override public void run ( ) { while ( true ) { int prefetchBatchSize ; boolean nackQueueMessages = false ; List < Message > messages = null ; try { if ( isClosed ( ) ) { break ; } synchronized ( stateLock ) { waitForStart ( ) ; waitForPrefetch ( ) ; prefetchBatchSize = Math . min ( numberOfMessagesToFetch ( ) , SQSMessagingClientConstants . MAX_BATCH ) ; } if ( ! isClosed ( ) ) { messages = getMessages ( prefetchBatchSize ) ; } if ( messages != null && ! messages . isEmpty ( ) ) { processReceivedMessages ( messages ) ; } } catch ( InterruptedException e ) { nackQueueMessages = true ; break ; } catch ( Throwable e ) { LOG . error ( "Unexpected exception when prefetch messages:" , e ) ; nackQueueMessages = true ; throw new RuntimeException ( e ) ; } finally { if ( isClosed ( ) || nackQueueMessages ) { nackQueueMessages ( ) ; } } } }
Runs until the message consumer is closed and in-progress SQS
<code>receiveMessage</code> call returns.
<P>
This blocks if configured number of prefetched messages are already
received or connection has not started yet.
<P>
After consumer is closed, all the messages inside internal buffer will
be negatively acknowledged.