@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();
        long 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 broadcastItem
                    ? collector.offerBroadcast(broadcastItem)
                    : 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();
    }
@Test
    public void when_receiveTwoObjects_then_emitThem() throws IOException {
        pushObjects(1, 2);
        t.call();
        assertEquals(asList(1, 2), collector.getBuffer());
    }