FireAndForgetRequesterMono extends Mono<Void> implements Subscription, Scannable { @Override public void subscribe(CoreSubscriber<? super Void> actual) { long previousState = markSubscribed(STATE, this); if (isSubscribedOrTerminated(previousState)) { Operators.error( actual, new IllegalStateException("FireAndForgetMono allows only a single Subscriber")); return; } actual.onSubscribe(this); final Payload p = this.payload; int mtu = this.mtu; try { if (!isValid(mtu, this.maxFrameLength, p, false)) { lazyTerminate(STATE, this); p.release(); actual.onError( new IllegalArgumentException( String.format(INVALID_PAYLOAD_ERROR_MESSAGE, this.maxFrameLength))); return; } } catch (IllegalReferenceCountException e) { lazyTerminate(STATE, this); actual.onError(e); return; } final int streamId; try { streamId = this.requesterResponderSupport.getNextStreamId(); } catch (Throwable t) { lazyTerminate(STATE, this); p.release(); actual.onError(Exceptions.unwrap(t)); return; } try { if (isTerminated(this.state)) { p.release(); return; } sendReleasingPayload( streamId, FrameType.REQUEST_FNF, mtu, p, this.connection, this.allocator, true); } catch (Throwable e) { lazyTerminate(STATE, this); actual.onError(e); return; } lazyTerminate(STATE, this); actual.onComplete(); } FireAndForgetRequesterMono(Payload payload, RequesterResponderSupport requesterResponderSupport); @Override void subscribe(CoreSubscriber<? super Void> actual); @Override void request(long n); @Override void cancel(); @Override @Nullable Void block(Duration m); @Override @Nullable Void block(); @Override Object scanUnsafe(Scannable.Attr key); @Override @NonNull String stepName();  }
@Test public void shouldSubscribeExactlyOnce1() { final TestRequesterResponderSupport streamManager = TestRequesterResponderSupport.client(); final LeaksTrackingByteBufAllocator allocator = streamManager.getAllocator(); final TestDuplexConnection sender = streamManager.getDuplexConnection(); for (int i = 1; i < 50000; i += 2) { final Payload payload = ByteBufPayload.create("testData", "testMetadata"); final FireAndForgetRequesterMono fireAndForgetRequesterMono = new FireAndForgetRequesterMono(payload, streamManager); final StateAssert<FireAndForgetRequesterMono> stateAssert = StateAssert.assertThat(FireAndForgetRequesterMono.STATE, fireAndForgetRequesterMono); Assertions.assertThatThrownBy( () -> RaceTestUtils.race( () -> { AtomicReference<Throwable> atomicReference = new AtomicReference(); fireAndForgetRequesterMono.subscribe(null, atomicReference::set); Throwable throwable = atomicReference.get(); if (throwable != null) { throw Exceptions.propagate(throwable); } }, fireAndForgetRequesterMono::block)) .matches( t -> { Assertions.assertThat(t) .hasMessageContaining("FireAndForgetMono allows only a single Subscriber"); return true; }); final ByteBuf frame = sender.awaitFrame(); FrameAssert.assertThat(frame) .isNotNull() .hasPayloadSize( "testData".getBytes(CharsetUtil.UTF_8).length + "testMetadata".getBytes(CharsetUtil.UTF_8).length) .hasMetadata("testMetadata") .hasData("testData") .hasNoFragmentsFollow() .typeOf(FrameType.REQUEST_FNF) .hasClientSideStreamId() .hasStreamId(i) .hasNoLeaks(); stateAssert.isTerminated(); streamManager.assertNoActiveStreams(); } Assertions.assertThat(sender.isEmpty()).isTrue(); allocator.assertHasNoLeaks(); }