public int doWork()
    {
        final long nowNs = nanoClock.nanoTime();
        cachedNanoClock.update(nowNs);
        dutyCycleTracker.measureAndUpdate(nowNs);

        int workCount = commandQueue.drain(CommandProxy.RUN_TASK, Configuration.COMMAND_DRAIN_LIMIT);

        final int bytesReceived = dataTransportPoller.pollTransports();
        totalBytesReceived.getAndAddOrdered(bytesReceived);

        final PublicationImage[] publicationImages = this.publicationImages;
        for (int lastIndex = publicationImages.length - 1, i = lastIndex; i >= 0; i--)
        {
            final PublicationImage image = publicationImages[i];
            if (image.isConnected(nowNs))
            {
                image.checkEosForDrainTransition(nowNs);

                workCount += image.sendPendingStatusMessage(nowNs);
                workCount += image.processPendingLoss();
                workCount += image.initiateAnyRttMeasurements(nowNs);
            }
            else
            {
                this.publicationImages = 1 == this.publicationImages.length ?
                    EMPTY_IMAGES : ArrayUtil.remove(this.publicationImages, i);
                image.removeFromDispatcher();
                image.receiverRelease();
            }
        }

        checkPendingSetupMessages(nowNs);

        if (reResolutionCheckIntervalNs > 0 && (reResolutionDeadlineNs - nowNs) < 0)
        {
            reResolutionDeadlineNs = nowNs + reResolutionCheckIntervalNs;
            dataTransportPoller.checkForReResolutions(nowNs, conductorProxy);
        }

        return workCount + bytesReceived;
    }
@Test
    @InterruptAfter(10)
    void shouldCreateRcvTermAndSendSmOnSetup() throws IOException
    {
        receiverProxy.registerReceiveChannelEndpoint(receiveChannelEndpoint);
        receiverProxy.addSubscription(receiveChannelEndpoint, STREAM_ID);

        receiver.doWork();
        receiver.doWork();

        fillSetupFrame(setupHeader);
        receiveChannelEndpoint.onSetupMessage(
            setupHeader, setupBuffer, SetupFlyweight.HEADER_LENGTH, senderAddress, 0);

        final PublicationImage image = new PublicationImage(
            CORRELATION_ID,
            ctx,
            receiveChannelEndpoint,
            0,
            senderAddress,
            SESSION_ID,
            STREAM_ID,
            INITIAL_TERM_ID,
            ACTIVE_TERM_ID,
            INITIAL_TERM_OFFSET,
            (short)0,
            rawLog,
            mockFeedbackDelayGenerator,
            POSITIONS,
            mockHighestReceivedPosition,
            mockRebuildPosition,
            SOURCE_IDENTITY,
            congestionControl);

        final int messagesRead = drainConductorQueue(
            (e) ->
            {
                // pass in new term buffer from conductor, which should trigger SM
                receiverProxy.newPublicationImage(receiveChannelEndpoint, image);
            });

        assertThat(messagesRead, is(1));

        nanoClock.advance(STATUS_MESSAGE_TIMEOUT * 2);
        receiver.doWork();

        final ByteBuffer rcvBuffer = ByteBuffer.allocateDirect(256);
        InetSocketAddress rcvAddress;

        do
        {
            rcvAddress = (InetSocketAddress)senderChannel.receive(rcvBuffer);
        }
        while (null == rcvAddress);

        statusHeader.wrap(new UnsafeBuffer(rcvBuffer));

        assertNotNull(rcvAddress);
        assertThat(rcvAddress.getPort(), is(UDP_CHANNEL.remoteData().getPort()));
        assertThat(statusHeader.headerType(), is(HeaderFlyweight.HDR_TYPE_SM));
        assertThat(statusHeader.streamId(), is(STREAM_ID));
        assertThat(statusHeader.sessionId(), is(SESSION_ID));
        assertThat(statusHeader.consumptionTermId(), is(ACTIVE_TERM_ID));
        assertThat(statusHeader.frameLength(), is(StatusMessageFlyweight.HEADER_LENGTH));
    }