@Test public void cancel() { TestSubscriber<Integer> ts = new TestSubscriber<>(); SubmissionPublisher<Integer> sp1 = new SubmissionPublisher<>(Runnable::run, 32); SubmissionPublisher<Integer> sp2 = new SubmissionPublisher<>(Runnable::run, 32); Multi.create(sp1) .flatMap(v -> sp2) .subscribe(ts); assertThat(sp1.hasSubscribers(), is(true)); assertThat(sp2.hasSubscribers(), is(false)); sp1.submit(1); assertThat(sp1.hasSubscribers(), is(true)); assertThat(sp2.hasSubscribers(), is(true)); ts.getSubcription().cancel(); assertThat(sp1.hasSubscribers(), is(false)); assertThat(sp2.hasSubscribers(), is(false)); }
@Override public void subscribe(Flow.Subscriber<? super R> subscriber) { Objects.requireNonNull(subscriber, "subscriber is null"); source.subscribe(new FlatMapSubscriber<>(subscriber, mapper, maxConcurrency, prefetch, delayErrors)); }
MultiFlatMapPublisher implements Multi<R> { @Override public void subscribe(Flow.Subscriber<? super R> subscriber) { Objects.requireNonNull(subscriber, "subscriber is null"); source.subscribe(new FlatMapSubscriber<>(subscriber, mapper, maxConcurrency, prefetch, delayErrors)); } }
MultiFlatMapPublisher implements Multi<R> { @Override public void subscribe(Flow.Subscriber<? super R> subscriber) { Objects.requireNonNull(subscriber, "subscriber is null"); source.subscribe(new FlatMapSubscriber<>(subscriber, mapper, maxConcurrency, prefetch, delayErrors)); } MultiFlatMapPublisher(Multi<T> source,
                          Function<? super T, ? extends Flow.Publisher<? extends R>> mapper,
                          long maxConcurrency, long prefetch, boolean delayErrors); }
MultiFlatMapPublisher implements Multi<R> { @Override public void subscribe(Flow.Subscriber<? super R> subscriber) { Objects.requireNonNull(subscriber, "subscriber is null"); source.subscribe(new FlatMapSubscriber<>(subscriber, mapper, maxConcurrency, prefetch, delayErrors)); } MultiFlatMapPublisher(Multi<T> source,
                          Function<? super T, ? extends Flow.Publisher<? extends R>> mapper,
                          long maxConcurrency, long prefetch, boolean delayErrors); @Override void subscribe(Flow.Subscriber<? super R> subscriber); }
MultiFlatMapPublisher implements Multi<R> { @Override public void subscribe(Flow.Subscriber<? super R> subscriber) { Objects.requireNonNull(subscriber, "subscriber is null"); source.subscribe(new FlatMapSubscriber<>(subscriber, mapper, maxConcurrency, prefetch, delayErrors)); } MultiFlatMapPublisher(Multi<T> source,
                          Function<? super T, ? extends Flow.Publisher<? extends R>> mapper,
                          long maxConcurrency, long prefetch, boolean delayErrors); @Override void subscribe(Flow.Subscriber<? super R> subscriber);  }