@Override
        public void run() {
            if (timeoutMode == StreamTimeoutMode.UNTIL_NEXT) {
                final long currentTimeNanos = System.nanoTime();
                final long elapsedNanos = currentTimeNanos - lastEventTimeNanos;

                if (elapsedNanos < timeoutNanos) {
                    final long delayNanos = timeoutNanos - elapsedNanos;
                    timeoutFuture = scheduleTimeout(delayNanos);
                    return;
                }
            }
            completed = true;
            delegate.onError(new StreamTimeoutException(
                    String.format(TIMEOUT_MESSAGE, timeoutDuration.toMillis(), timeoutMode)));
            assert subscription != null;
            subscription.cancel();
        }
Subscribes the given subscriber to this stream with timeout logic applied.

@param subscriber the subscriber to this stream
@param executor the executor for running timeout tasks and stream operations
@param options subscription options
@see StreamMessage#subscribe(Subscriber, EventExecutor, SubscriptionOption...)
core/src/main/java/com/linecorp/armeria/common/stream/TimeoutStreamMessage.java