@@ -371,41 +371,31 @@ public void processReceivedMessages(List<ReceivedMessage> messages, Runnable don
371371 processOutstandingBatches ();
372372 }
373373
374- public void processOutstandingBatches () {
375- while (true ) {
376- boolean batchDone = false ;
377- Runnable batchCallback = null ;
378- OutstandingMessage outstandingMessage ;
379- synchronized (outstandingMessageBatches ) {
380- OutstandingMessageBatch nextBatch = outstandingMessageBatches .peek ();
381- if (nextBatch == null ) {
382- return ;
383- }
384- outstandingMessage = nextBatch .messages .peek ();
385- if (outstandingMessage == null ) {
386- return ;
387- }
388- try {
389- // This is a non-blocking flow controller.
390- flowController .reserve (
391- 1 , outstandingMessage .receivedMessage ().getMessage ().getSerializedSize ());
392- } catch (FlowController .MaxOutstandingElementCountReachedException
393- | FlowController .MaxOutstandingRequestBytesReachedException flowControlException ) {
394- return ;
395- } catch (FlowControlException unexpectedException ) {
396- throw new IllegalStateException ("Flow control unexpected exception" , unexpectedException );
397- }
398- nextBatch .messages .poll (); // We got a hold to the message already.
399- batchDone = nextBatch .messages .isEmpty ();
400- if (batchDone ) {
401- outstandingMessageBatches .poll ();
402- batchCallback = nextBatch .doneCallback ;
374+ private void processOutstandingBatches () {
375+ synchronized (outstandingMessageBatches ) {
376+ for (OutstandingMessageBatch nextBatch = outstandingMessageBatches .poll ();
377+ nextBatch != null ;
378+ nextBatch = outstandingMessageBatches .poll ()) {
379+ for (OutstandingMessage nextMessage = nextBatch .messages .poll ();
380+ nextMessage != null ;
381+ nextMessage = nextBatch .messages .poll ()) {
382+ try {
383+ // This is a non-blocking flow controller.
384+ flowController .reserve (1 , nextMessage .receivedMessage .getMessage ().getSerializedSize ());
385+ } catch (FlowController .MaxOutstandingElementCountReachedException
386+ | FlowController .MaxOutstandingRequestBytesReachedException flowControlException ) {
387+ // Unwind previous changes in the batches outstanding.
388+ nextBatch .messages .addFirst (nextMessage );
389+ outstandingMessageBatches .addFirst (nextBatch );
390+ return ;
391+ } catch (FlowControlException unexpectedException ) {
392+ throw new IllegalStateException (
393+ "Flow control unexpected exception" , unexpectedException );
394+ }
395+ processOutstandingMessage (
396+ nextMessage .receivedMessage .getMessage (), nextMessage .ackHandler );
403397 }
404- }
405- processOutstandingMessage (
406- outstandingMessage .receivedMessage .getMessage (), outstandingMessage .ackHandler );
407- if (batchDone ) {
408- batchCallback .run ();
398+ nextBatch .doneCallback .run ();
409399 }
410400 }
411401 }
0 commit comments