@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++) {
		//use an extra handler
		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();
	}
}