ReceiverTasklet implements Tasklet { @Override @Nonnull public ProgressState call() { if (receptionDone) { return collector.offerBroadcast(DONE_ITEM); } if (connectionChanged) { throw new RestartableException("The member was reconnected: " + sourceAddressString); } tracker.reset(); tracker.notDone(); tryFillInbox(); int ackItemLocal = 0; for (ObjWithPtionIdAndSize o; (o = inbox.peek()) != null; ) { final Object item = o.getItem(); if (item == DONE_ITEM) { receptionDone = true; inbox.remove(); assert inbox.peek() == null : "Found something in the queue beyond the DONE_ITEM: " + inbox.remove(); break; } ProgressState outcome = item instanceof BroadcastItem ? collector.offerBroadcast((BroadcastItem) item) : collector.offer(item, o.getPartitionId()); if (!outcome.isDone()) { tracker.madeProgress(outcome.isMadeProgress()); break; } tracker.madeProgress(); inbox.remove(); ackItemLocal += o.estimatedMemoryFootprint; } ackItem(ackItemLocal); numWaitingInInbox = inbox.size(); return tracker.toProgressState(); } ReceiverTasklet(
            OutboundCollector collector, InternalSerializationService serializationService,
            int rwinMultiplier, int flowControlPeriodMs, LoggingService loggingService,
            Address sourceAddress, int ordinal, String destinationVertexName,
            Connection memberConnection
    ); @Override @Nonnull ProgressState call(); int updateAndGetSendSeqLimitCompressed(Connection expectedConnection); @Override String toString(); @Override void provideDynamicMetrics(MetricDescriptor descriptor, MetricsCollectionContext context);  }
@Test public void when_receiveTwoObjects_then_emitThem() throws IOException { pushObjects(1, 2); t.call(); assertEquals(asList(1, 2), collector.getBuffer()); }