MonoSendMany extends MonoSend<I, O> implements Scannable { static MonoSendMany<ByteBuf, ByteBuf> byteBufSource(Publisher<? extends ByteBuf> source, Channel channel, Predicate<ByteBuf> predicate) { return new MonoSendMany<>(source, channel, predicate, TRANSFORMATION_FUNCTION_BB, CONSUMER_NOCHECK_CLEANUP, SIZE_OF_BB); } 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 shouldNotLeakOnRacingCancelAndOnNext() { int messagesToSend = 128; for (int i = 0; i < 10000; i++) { EmbeddedChannel channel = new EmbeddedChannel(true, true, new ChannelHandlerAdapter() {}); TestPublisher<ByteBuf> source = TestPublisher.createNoncompliant(TestPublisher.Violation.DEFER_CANCELLATION); IdentityHashMap<ReferenceCounted, Object> discarded = new IdentityHashMap<>(); MonoSendMany<ByteBuf, ByteBuf> m = MonoSendMany.byteBufSource(source, channel, b -> false); BaseSubscriber<Void> testSubscriber = m .doOnDiscard(ReferenceCounted.class, v -> discarded.put(v, null)) .subscribeWith(new BaseSubscriber<Void>() {}); Queue<Object> messages = channel.outboundMessages(); Queue<ByteBuf> buffersToSend = new ArrayDeque<>(messagesToSend); for (int j = 0; j < messagesToSend; j++) { buffersToSend.offer(ByteBufAllocator.DEFAULT.buffer().writeInt(j)); } RaceTestUtils.race(testSubscriber::cancel, () -> { for (ByteBuf buf : buffersToSend) { source.next(buf); } }); channel.flush(); messages.forEach(ReferenceCountUtil::safeRelease); assertThat(discarded.size() + messages.size()) .as("Expect all element are flushed or discarded but was discarded " + ": [" + discarded.size() + "], flushed : [" + messages.size() + "]") .isEqualTo(messagesToSend); } }
@Test @SuppressWarnings("deprecation") public void shouldNotLeakIfFusedOnRacingCancelAndOnNext() { int messagesToSend = 128; ArrayBlockingQueue<ReferenceCounted> discarded = new ArrayBlockingQueue<>(messagesToSend * 2); Hooks.onNextDropped(v -> { ReferenceCountUtil.safeRelease(v); discarded.add((ReferenceCounted)v); }); for (int i = 0; i < 10000; i++) { EmbeddedChannel channel = new EmbeddedChannel(true, true, new ChannelHandlerAdapter() {}); UnicastProcessor<ByteBuf> source = UnicastProcessor.create(); MonoSendMany<ByteBuf, ByteBuf> m = MonoSendMany.byteBufSource(source, channel, b -> false); BaseSubscriber<Void> testSubscriber = m .doOnDiscard(ReferenceCounted.class, discarded::add) .subscribeWith(new BaseSubscriber<Void>() {}); Queue<Object> messages = channel.outboundMessages(); Queue<ByteBuf> buffersToSend = new ArrayDeque<>(messagesToSend); for (int j = 0; j < messagesToSend; j++) { buffersToSend.offer(ByteBufAllocator.DEFAULT.buffer().writeInt(j)); } RaceTestUtils.race(testSubscriber::cancel, () -> { for (ByteBuf buf : buffersToSend) { source.onNext(buf); } }); IdentityHashMap<ReferenceCounted, ?> distinctDiscarded = discarded.stream().collect(Collectors.toMap(Function.identity(), Function.identity(), (r1, r2) -> r1, IdentityHashMap::new)); channel.flush(); messages.forEach(ReferenceCountUtil::release); assertThat(distinctDiscarded.size() + messages.size()) .as("Expect all element are flushed or discarded but was discarded " + ": [" + distinctDiscarded.size() + "], flushed : [" + messages.size() + "]") .isEqualTo(messagesToSend); discarded.clear(); } }