@Test public void shouldBecomeReplaySlowConsumer() { final long position = 0; channelWillWrite(0); onReplayMessage(0, position); assertBytesInBuffer(BODY_LENGTH); }
Action onReplayMessage( final DirectBuffer directBuffer, final int offset, final int bodyLength, final long timeInMs, final long position) { if (!isSlowConsumer()) { replayPaused = true; } attemptFramedMessage(directBuffer, offset, bodyLength, timeInMs, position, replayTracker); return CONTINUE; }
SenderEndPoint { Action onReplayMessage( final DirectBuffer directBuffer, final int offset, final int bodyLength, final long timeInMs, final long position) { if (!isSlowConsumer()) { replayPaused = true; } attemptFramedMessage(directBuffer, offset, bodyLength, timeInMs, position, replayTracker); return CONTINUE; } }
SenderEndPoint { Action onReplayMessage( final DirectBuffer directBuffer, final int offset, final int bodyLength, final long timeInMs, final long position) { if (!isSlowConsumer()) { replayPaused = true; } attemptFramedMessage(directBuffer, offset, bodyLength, timeInMs, position, replayTracker); return CONTINUE; } SenderEndPoint(
        final long connectionId,
        final int libraryId,
        final BlockablePosition outboundBlockablePosition,
        final BlockablePosition replayBlockablePosition,
        final TcpChannel channel,
        final AtomicCounter bytesInBuffer,
        final AtomicCounter invalidLibraryAttempts,
        final ErrorHandler errorHandler,
        final Framer framer,
        final int maxBytesInBuffer,
        final long slowConsumerTimeoutInMs,
        final long timeInMs,
        final SenderSequenceNumber senderSequenceNumber,
        final MessageTimingHandler messageTimingHandler); }
SenderEndPoint { Action onReplayMessage( final DirectBuffer directBuffer, final int offset, final int bodyLength, final long timeInMs, final long position) { if (!isSlowConsumer()) { replayPaused = true; } attemptFramedMessage(directBuffer, offset, bodyLength, timeInMs, position, replayTracker); return CONTINUE; } SenderEndPoint(
        final long connectionId,
        final int libraryId,
        final BlockablePosition outboundBlockablePosition,
        final BlockablePosition replayBlockablePosition,
        final TcpChannel channel,
        final AtomicCounter bytesInBuffer,
        final AtomicCounter invalidLibraryAttempts,
        final ErrorHandler errorHandler,
        final Framer framer,
        final int maxBytesInBuffer,
        final long slowConsumerTimeoutInMs,
        final long timeInMs,
        final SenderSequenceNumber senderSequenceNumber,
        final MessageTimingHandler messageTimingHandler); long connectionId(); void libraryId(final int libraryId, final BlockablePosition blockablePosition); int libraryId(); void close(); }
SenderEndPoint { Action onReplayMessage( final DirectBuffer directBuffer, final int offset, final int bodyLength, final long timeInMs, final long position) { if (!isSlowConsumer()) { replayPaused = true; } attemptFramedMessage(directBuffer, offset, bodyLength, timeInMs, position, replayTracker); return CONTINUE; } SenderEndPoint(
        final long connectionId,
        final int libraryId,
        final BlockablePosition outboundBlockablePosition,
        final BlockablePosition replayBlockablePosition,
        final TcpChannel channel,
        final AtomicCounter bytesInBuffer,
        final AtomicCounter invalidLibraryAttempts,
        final ErrorHandler errorHandler,
        final Framer framer,
        final int maxBytesInBuffer,
        final long slowConsumerTimeoutInMs,
        final long timeInMs,
        final SenderSequenceNumber senderSequenceNumber,
        final MessageTimingHandler messageTimingHandler); long connectionId(); void libraryId(final int libraryId, final BlockablePosition blockablePosition); int libraryId(); void close();  }