@Test
public void shouldNotLeakOnRacingCancelAndOnNext() {

	int messagesToSend = 128;
	for (int i = 0; i < 10000; i++) {
		//use an extra handler
		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);
	}
}