MonoSendMany extends MonoSend<I, O> implements Scannable { static MonoSendMany<?, ?> objectSource(Publisher<?> source, Channel channel, Predicate<Object> predicate) { return new MonoSendMany<>(source, channel, predicate, TRANSFORMATION_FUNCTION, CONSUMER_NOCHECK_CLEANUP, SIZE_OF); } MonoSendMany(Publisher<? extends I> source,
			Channel channel,
			Predicate<I> predicate,
			Function<? super I, ? extends O> transformer,
			Consumer<? super I> sourceCleanup,
			ToIntFunction<O> sizeOf); @Override void subscribe(CoreSubscriber<? super Void> destination); @Override @Nullable @SuppressWarnings("rawtypes") Object scanUnsafe(Attr key);  }
@Test public void cleanupCancelCloseFuture() { EmbeddedChannel channel = new EmbeddedChannel(new ChannelHandlerAdapter() {}); Mono<Void> m = MonoSendMany.objectSource(Flux.fromArray(new String[]{"test", "test2"}).concatWith(Mono.never()), channel, b -> false); List<WeakReference<Subscription>> _w = new ArrayList<>(1); StepVerifier.create(m) .consumeSubscriptionWith(s -> _w.add(new WeakReference<>(s))) .then(channel::runPendingTasks) .thenCancel() .verify(); System.gc(); wait(_w.get(0)); }
@Test public void cleanupErrorCloseFuture() { EmbeddedChannel channel = new EmbeddedChannel(new ChannelHandlerAdapter() {}); Mono<Void> m = MonoSendMany.objectSource(Mono.error(new Exception("boo")), channel, b -> false); List<WeakReference<Subscription>> _w = new ArrayList<>(1); StepVerifier.create(m) .consumeSubscriptionWith(s -> _w.add(new WeakReference<>(s))) .then(channel::runPendingTasks) .verifyErrorMessage("boo"); System.gc(); wait(_w.get(0)); }
@Test public void testPromiseSendTimeout() { EmbeddedChannel channel = new EmbeddedChannel(new WriteTimeoutHandler(1), new ChannelHandlerAdapter() {}); Flux<String> flux = Flux.range(0, 257).map(count -> count + ""); Mono<Void> m = MonoSendMany.objectSource(flux, channel, b -> false); StepVerifier.create(m) .then(() -> { channel.runPendingTasks(); for (int i = 0; i < 257; i++) { assertThat(channel.<String>readOutbound()).isEqualTo(i + ""); } }) .verifyComplete(); }
@Test public void cleanupFuseableSyncCloseFuture() { EmbeddedChannel channel = new EmbeddedChannel(new ChannelHandlerAdapter() {}); Mono<Void> m = MonoSendMany.objectSource(Flux.fromArray(new String[]{"test", "test2"}), channel, b -> false); List<WeakReference<Subscription>> _w = new ArrayList<>(1); StepVerifier.create(m) .consumeSubscriptionWith(s -> _w.add(new WeakReference<>(s))) .then(() -> { channel.runPendingTasks(); assertThat(channel.<String>readOutbound()).isEqualToIgnoringCase("test"); assertThat(channel.<String>readOutbound()).isEqualToIgnoringCase("test2"); }) .verifyComplete(); System.gc(); wait(_w.get(0)); }
@Test public void cleanupFuseableAsyncCloseFuture() { EmbeddedChannel channel = new EmbeddedChannel(new ChannelHandlerAdapter() {}); Mono<Void> m = MonoSendMany.objectSource(Flux.fromArray(new String[]{"test", "test2"}).limitRate(10), channel, b -> false); List<WeakReference<Subscription>> _w = new ArrayList<>(1); StepVerifier.create(m) .consumeSubscriptionWith(s -> _w.add(new WeakReference<>(s))) .then(() -> { channel.runPendingTasks(); assertThat(channel.<String>readOutbound()).isEqualToIgnoringCase("test"); assertThat(channel.<String>readOutbound()).isEqualToIgnoringCase("test2"); }) .verifyComplete(); System.gc(); wait(_w.get(0)); }
@Test public void cleanupFuseableErrorCloseFuture() { EmbeddedChannel channel = new EmbeddedChannel(new ChannelHandlerAdapter() {}); Mono<Void> m = MonoSendMany.objectSource(Flux.fromArray(new String[]{"test", "test2"}).concatWith(Mono.error(new Exception("boo"))).limitRate(10), channel, b -> false); List<WeakReference<Subscription>> _w = new ArrayList<>(1); StepVerifier.create(m) .consumeSubscriptionWith(s -> _w.add(new WeakReference<>(s))) .then(() -> { channel.runPendingTasks(); assertThat(channel.<String>readOutbound()).isEqualToIgnoringCase("test"); assertThat(channel.<String>readOutbound()).isEqualToIgnoringCase("test2"); }) .verifyErrorMessage("boo"); System.gc(); wait(_w.get(0)); }