DelayCancelPublisher.java

import org.reactivestreams.Publisher;
import org.reactivestreams.Subscriber;
import org.reactivestreams.Subscription;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

import java.time.Duration;
import java.util.concurrent.atomic.AtomicBoolean;

public class DelayCancelPublisher<T> implements Publisher<T> {

    private final Publisher<T> upstream;
    private final Mono<Long> delay;

    public DelayCancelPublisher(Publisher<T> upstream, Duration duration) {
        this.upstream = upstream;
        this.delay = Mono.delay(duration);
    }

    public static <T> Flux<T> delayCancel(Publisher<T> upstream, Duration duration) {
        return Flux.from(new DelayCancelPublisher<>(upstream, duration));
    }

    @Override
    public void subscribe(Subscriber<? super T> s) {
        upstream.subscribe(new DelayCancelSubscriber<>(s, delay));
    }

    private static class DelayCancelSubscriber<T> implements Subscriber<T> {
        private final Subscriber<? super T> subscriber;
        private final Mono<Long> delay;
        private final AtomicBoolean cancelled = new AtomicBoolean();

        public DelayCancelSubscriber(Subscriber<? super T> subscriber, Mono<Long> delay) {
            this.subscriber = subscriber;
            this.delay = delay;
        }

        @Override
        public void onSubscribe(Subscription upstream) {
            subscriber.onSubscribe(new Subscription() {
                @Override
                public void request(long n) {
                    upstream.request(n);
                }

                @Override
                public void cancel() {
                    if (cancelled.compareAndSet(false, true)) {
                        delay.subscribe(v -> upstream.cancel());
                    }
                }
            });
        }

        @Override
        public void onNext(T t) {
            if (!cancelled.get()) {
                subscriber.onNext(t);
            }
        }

        @Override
        public void onError(Throwable t) {
            subscriber.onError(t);
        }

        @Override
        public void onComplete() {
            subscriber.onComplete();
        }
    }
}