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
    void shouldNotOverwriteDataFrameWithHeartbeat()
    {
        receiverProxy.registerReceiveChannelEndpoint(receiveChannelEndpoint);
        receiverProxy.addSubscription(receiveChannelEndpoint, STREAM_ID);

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

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

        final int commandsRead = drainConductorQueue(
            (e) ->
            {
                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);

                receiverProxy.newPublicationImage(receiveChannelEndpoint, image);
            });

        assertThat(commandsRead, is(1));

        receiver.doWork();

        fillDataFrame(dataHeader, 0);  // initial data frame
        receiveChannelEndpoint.onDataPacket(dataHeader, dataBuffer, dataHeader.frameLength(), senderAddress, 0);

        fillDataFrame(dataHeader, 0);  // heartbeat with same term offset
        receiveChannelEndpoint.onDataPacket(dataHeader, dataBuffer, dataHeader.frameLength(), senderAddress, 0);

        final int readOutcome = TermReader.read(
            termBuffers[ACTIVE_INDEX],
            INITIAL_TERM_OFFSET,
            (buffer, offset, length, header) ->
            {
                assertThat(header.type(), is(HeaderFlyweight.HDR_TYPE_DATA));
                assertThat(header.termId(), is(ACTIVE_TERM_ID));
                assertThat(header.streamId(), is(STREAM_ID));
                assertThat(header.sessionId(), is(SESSION_ID));
                assertThat(header.termOffset(), is(0));
                assertThat(header.frameLength(), is(DataHeaderFlyweight.HEADER_LENGTH + FAKE_PAYLOAD.length));
            },
            Integer.MAX_VALUE,
            header,
            mockErrorHandler,
            0,
            mockSubscriberPosition);

        assertThat(readOutcome, is(1));
    }