Non blocking. Returns immediately if Q is empty. Returns number of elements consumed from Q.
(Consumer consumer, ExitCondition exitCond)
| 104 | * Non blocking. Returns immediately if Q is empty. Returns number of elements consumed from Q. |
| 105 | */ |
| 106 | private int consumeImpl(Consumer consumer, ExitCondition exitCond) throws InterruptedException { |
| 107 | int drainCount = 0; |
| 108 | while (exitCond.keepRunning()) { |
| 109 | Object tuple = recvQueue.poll(); |
| 110 | if (tuple == null) { |
| 111 | break; |
| 112 | } |
| 113 | consumer.accept(tuple); |
| 114 | ++drainCount; |
| 115 | } |
| 116 | |
| 117 | int overflowDrainCount = 0; |
| 118 | int limit = overflowQ.size(); |
| 119 | while (exitCond.keepRunning() && (overflowDrainCount < limit)) { // 2nd cond prevents staying stuck with consuming overflow |
| 120 | Object tuple = overflowQ.poll(); |
| 121 | ++overflowDrainCount; |
| 122 | consumer.accept(tuple); |
| 123 | } |
| 124 | int total = drainCount + overflowDrainCount; |
| 125 | if (total > 0) { |
| 126 | consumer.flush(); |
| 127 | } |
| 128 | return total; |
| 129 | } |
| 130 | |
| 131 | // Non Blocking. returns true/false indicating success/failure. Fails if full. |
| 132 | private boolean tryPublishInternal(Object obj) { |