()
| 469 | } |
| 470 | |
| 471 | @Override |
| 472 | @Nullable |
| 473 | public final synchronized InputStream next() { |
| 474 | InputStream stream = null; |
| 475 | if (firstMessage != null) { |
| 476 | stream = firstMessage; |
| 477 | firstMessage = null; |
| 478 | } else if (numRequestedMessages > 0 && messageAvailable()) { |
| 479 | stream = assembleNextMessage(); |
| 480 | } |
| 481 | if (stream != null) { |
| 482 | numRequestedMessages -= 1; |
| 483 | } else { |
| 484 | producingMessages = false; |
| 485 | if (receivedAllTransactions()) { |
| 486 | // That's the last of the messages delivered. |
| 487 | if (!isClosed()) { |
| 488 | onDeliveryState(State.ALL_MESSAGES_DELIVERED); |
| 489 | deliver(); |
| 490 | } |
| 491 | } |
| 492 | } |
| 493 | return stream; |
| 494 | } |
| 495 | |
| 496 | @GuardedBy("this") |
| 497 | private InputStream assembleNextMessage() { |
nothing calls this directly
no test coverage detected