@Test public void shouldNotDisconnectAtStartDueToTimeout() { final long timeInMs = DEFAULT_SLOW_CONSUMER_TIMEOUT_IN_MS - 1; endPoint.checkTimeouts(timeInMs); verifySlowConsumerDisconnect(never()); verifyNoMoreErrors(); verifyDoesNotBlockLibrary(); }
boolean checkTimeouts(final long timeInMs) { if (isSlowConsumer() && timeInMs > sendingTimeoutTimeInMs) { errorHandler.onError(new IllegalStateException(String.format( "Slow Consumer Disconnected conn=%d,sess=%d @ time %d, Due to not being able to write since %d", connectionId, sessionId, timeInMs, sendingTimeoutTimeInMs - slowConsumerTimeoutInMs))); removeEndpoint(SLOW_CONSUMER); return true; } return false; }
SenderEndPoint { boolean checkTimeouts(final long timeInMs) { if (isSlowConsumer() && timeInMs > sendingTimeoutTimeInMs) { errorHandler.onError(new IllegalStateException(String.format( "Slow Consumer Disconnected conn=%d,sess=%d @ time %d, Due to not being able to write since %d", connectionId, sessionId, timeInMs, sendingTimeoutTimeInMs - slowConsumerTimeoutInMs))); removeEndpoint(SLOW_CONSUMER); return true; } return false; } }
SenderEndPoint { boolean checkTimeouts(final long timeInMs) { if (isSlowConsumer() && timeInMs > sendingTimeoutTimeInMs) { errorHandler.onError(new IllegalStateException(String.format( "Slow Consumer Disconnected conn=%d,sess=%d @ time %d, Due to not being able to write since %d", connectionId, sessionId, timeInMs, sendingTimeoutTimeInMs - slowConsumerTimeoutInMs))); removeEndpoint(SLOW_CONSUMER); return true; } return false; } 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 { boolean checkTimeouts(final long timeInMs) { if (isSlowConsumer() && timeInMs > sendingTimeoutTimeInMs) { errorHandler.onError(new IllegalStateException(String.format( "Slow Consumer Disconnected conn=%d,sess=%d @ time %d, Due to not being able to write since %d", connectionId, sessionId, timeInMs, sendingTimeoutTimeInMs - slowConsumerTimeoutInMs))); removeEndpoint(SLOW_CONSUMER); return true; } return false; } 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 { boolean checkTimeouts(final long timeInMs) { if (isSlowConsumer() && timeInMs > sendingTimeoutTimeInMs) { errorHandler.onError(new IllegalStateException(String.format( "Slow Consumer Disconnected conn=%d,sess=%d @ time %d, Due to not being able to write since %d", connectionId, sessionId, timeInMs, sendingTimeoutTimeInMs - slowConsumerTimeoutInMs))); removeEndpoint(SLOW_CONSUMER); return true; } return false; } 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();  }