@Test public void testRun() { final ClientActorBehavior<U> behavior = mock(ClientActorBehavior.class); assertSame(behavior, connection.runTimer(behavior)); }
@VisibleForTesting final ClientActorBehavior<T> runTimer(final ClientActorBehavior<T> current) { lock.lock(); final List<ConnectionEntry> poisonEntries; final NoProgressException poisonCause; try { haveTimer = false; final long now = currentTime(); LOG.debug("{}: running timer on {}", context.persistenceId(), this); final long ticksSinceProgress = queue.ticksStalling(now); if (ticksSinceProgress < context.config().getNoProgressTimeout()) { final OptionalLong delay = lockedCheckTimeout(now); if (delay == null) { LOG.debug("{}: connection {} timed out", context.persistenceId(), this); return lockedReconnect(current, new RuntimeRequestException("Backend connection timed out", new TimeoutException())); } if (delay.isPresent()) { scheduleTimer(delay.getAsLong()); } else { LOG.debug("{}: not scheduling timeout on {}", context.persistenceId(), this); } return current; } LOG.error("Queue {} has not seen progress in {} seconds, failing all requests", this, TimeUnit.NANOSECONDS.toSeconds(ticksSinceProgress)); poisonCause = new NoProgressException(ticksSinceProgress); poisonEntries = lockedPoison(poisonCause); current.removeConnection(this); } finally { lock.unlock(); } poison(poisonEntries, poisonCause); return current; }
AbstractClientConnection { @VisibleForTesting final ClientActorBehavior<T> runTimer(final ClientActorBehavior<T> current) { lock.lock(); final List<ConnectionEntry> poisonEntries; final NoProgressException poisonCause; try { haveTimer = false; final long now = currentTime(); LOG.debug("{}: running timer on {}", context.persistenceId(), this); final long ticksSinceProgress = queue.ticksStalling(now); if (ticksSinceProgress < context.config().getNoProgressTimeout()) { final OptionalLong delay = lockedCheckTimeout(now); if (delay == null) { LOG.debug("{}: connection {} timed out", context.persistenceId(), this); return lockedReconnect(current, new RuntimeRequestException("Backend connection timed out", new TimeoutException())); } if (delay.isPresent()) { scheduleTimer(delay.getAsLong()); } else { LOG.debug("{}: not scheduling timeout on {}", context.persistenceId(), this); } return current; } LOG.error("Queue {} has not seen progress in {} seconds, failing all requests", this, TimeUnit.NANOSECONDS.toSeconds(ticksSinceProgress)); poisonCause = new NoProgressException(ticksSinceProgress); poisonEntries = lockedPoison(poisonCause); current.removeConnection(this); } finally { lock.unlock(); } poison(poisonEntries, poisonCause); return current; } }
AbstractClientConnection { @VisibleForTesting final ClientActorBehavior<T> runTimer(final ClientActorBehavior<T> current) { lock.lock(); final List<ConnectionEntry> poisonEntries; final NoProgressException poisonCause; try { haveTimer = false; final long now = currentTime(); LOG.debug("{}: running timer on {}", context.persistenceId(), this); final long ticksSinceProgress = queue.ticksStalling(now); if (ticksSinceProgress < context.config().getNoProgressTimeout()) { final OptionalLong delay = lockedCheckTimeout(now); if (delay == null) { LOG.debug("{}: connection {} timed out", context.persistenceId(), this); return lockedReconnect(current, new RuntimeRequestException("Backend connection timed out", new TimeoutException())); } if (delay.isPresent()) { scheduleTimer(delay.getAsLong()); } else { LOG.debug("{}: not scheduling timeout on {}", context.persistenceId(), this); } return current; } LOG.error("Queue {} has not seen progress in {} seconds, failing all requests", this, TimeUnit.NANOSECONDS.toSeconds(ticksSinceProgress)); poisonCause = new NoProgressException(ticksSinceProgress); poisonEntries = lockedPoison(poisonCause); current.removeConnection(this); } finally { lock.unlock(); } poison(poisonEntries, poisonCause); return current; } private  AbstractClientConnection(final AbstractClientConnection<T> oldConn, final TransmitQueue newQueue,
            final String backendName);   AbstractClientConnection(final ClientActorContext context, final Long cookie, final String backendName,
            final int queueDepth);   AbstractClientConnection(final AbstractClientConnection<T> oldConn);   AbstractClientConnection(final AbstractClientConnection<T> oldConn, final T newBackend,
            final int queueDepth); }
AbstractClientConnection { @VisibleForTesting final ClientActorBehavior<T> runTimer(final ClientActorBehavior<T> current) { lock.lock(); final List<ConnectionEntry> poisonEntries; final NoProgressException poisonCause; try { haveTimer = false; final long now = currentTime(); LOG.debug("{}: running timer on {}", context.persistenceId(), this); final long ticksSinceProgress = queue.ticksStalling(now); if (ticksSinceProgress < context.config().getNoProgressTimeout()) { final OptionalLong delay = lockedCheckTimeout(now); if (delay == null) { LOG.debug("{}: connection {} timed out", context.persistenceId(), this); return lockedReconnect(current, new RuntimeRequestException("Backend connection timed out", new TimeoutException())); } if (delay.isPresent()) { scheduleTimer(delay.getAsLong()); } else { LOG.debug("{}: not scheduling timeout on {}", context.persistenceId(), this); } return current; } LOG.error("Queue {} has not seen progress in {} seconds, failing all requests", this, TimeUnit.NANOSECONDS.toSeconds(ticksSinceProgress)); poisonCause = new NoProgressException(ticksSinceProgress); poisonEntries = lockedPoison(poisonCause); current.removeConnection(this); } finally { lock.unlock(); } poison(poisonEntries, poisonCause); return current; } private  AbstractClientConnection(final AbstractClientConnection<T> oldConn, final TransmitQueue newQueue,
            final String backendName);   AbstractClientConnection(final ClientActorContext context, final Long cookie, final String backendName,
            final int queueDepth);   AbstractClientConnection(final AbstractClientConnection<T> oldConn);   AbstractClientConnection(final AbstractClientConnection<T> oldConn, final T newBackend,
            final int queueDepth); final ClientActorContext context(); final @NonNull Long cookie(); final ActorRef localActor(); final long currentTime(); final void sendRequest(final Request<?, ?> request, final Consumer<Response<?, ?>> callback); final void enqueueRequest(final Request<?, ?> request, final Consumer<Response<?, ?>> callback,
            final long enqueuedTicks); final void enqueueEntry(final ConnectionEntry entry, final long now); abstract Optional<T> getBackendInfo(); @Override final String toString(); }
AbstractClientConnection { @VisibleForTesting final ClientActorBehavior<T> runTimer(final ClientActorBehavior<T> current) { lock.lock(); final List<ConnectionEntry> poisonEntries; final NoProgressException poisonCause; try { haveTimer = false; final long now = currentTime(); LOG.debug("{}: running timer on {}", context.persistenceId(), this); final long ticksSinceProgress = queue.ticksStalling(now); if (ticksSinceProgress < context.config().getNoProgressTimeout()) { final OptionalLong delay = lockedCheckTimeout(now); if (delay == null) { LOG.debug("{}: connection {} timed out", context.persistenceId(), this); return lockedReconnect(current, new RuntimeRequestException("Backend connection timed out", new TimeoutException())); } if (delay.isPresent()) { scheduleTimer(delay.getAsLong()); } else { LOG.debug("{}: not scheduling timeout on {}", context.persistenceId(), this); } return current; } LOG.error("Queue {} has not seen progress in {} seconds, failing all requests", this, TimeUnit.NANOSECONDS.toSeconds(ticksSinceProgress)); poisonCause = new NoProgressException(ticksSinceProgress); poisonEntries = lockedPoison(poisonCause); current.removeConnection(this); } finally { lock.unlock(); } poison(poisonEntries, poisonCause); return current; } private  AbstractClientConnection(final AbstractClientConnection<T> oldConn, final TransmitQueue newQueue,
            final String backendName);   AbstractClientConnection(final ClientActorContext context, final Long cookie, final String backendName,
            final int queueDepth);   AbstractClientConnection(final AbstractClientConnection<T> oldConn);   AbstractClientConnection(final AbstractClientConnection<T> oldConn, final T newBackend,
            final int queueDepth); final ClientActorContext context(); final @NonNull Long cookie(); final ActorRef localActor(); final long currentTime(); final void sendRequest(final Request<?, ?> request, final Consumer<Response<?, ?>> callback); final void enqueueRequest(final Request<?, ?> request, final Consumer<Response<?, ?>> callback,
            final long enqueuedTicks); final void enqueueEntry(final ConnectionEntry entry, final long now); abstract Optional<T> getBackendInfo(); @Override final String toString(); static final long DEFAULT_BACKEND_ALIVE_TIMEOUT_NANOS; static final long DEFAULT_REQUEST_TIMEOUT_NANOS; static final long DEFAULT_NO_PROGRESS_TIMEOUT_NANOS; }