RSocketResponder does not remove Subscriber from sendingSubscriptions when response is empty

[hadix-lin]

io.rsocket.core.RSocketResponder#handleRequestResponse

private void handleRequestResponse(int streamId, Mono<Payload> response) {
    final BaseSubscriber<Payload> subscriber =
        new BaseSubscriber<Payload>() {
          private boolean isEmpty = true;

          @Override
          protected void hookOnNext(Payload payload) {
            if (isEmpty) {
              isEmpty = false;
            }

            if (!PayloadValidationUtils.isValid(mtu, payload, maxFrameLength)) {
              payload.release();
              cancel();
              final IllegalArgumentException t =
                  new IllegalArgumentException(INVALID_PAYLOAD_ERROR_MESSAGE);
              handleError(streamId, t);
              return;
            }

            ByteBuf byteBuf =
                PayloadFrameCodec.encodeNextCompleteReleasingPayload(allocator, streamId, payload);
            sendProcessor.onNext(byteBuf);
          }

          @Override
          protected void hookOnError(Throwable throwable) {
            if (sendingSubscriptions.remove(streamId, this)) {
              handleError(streamId, throwable);
            }
          }

          @Override
          protected void hookOnComplete() {
            if (isEmpty) {
              if (sendingSubscriptions.remove(streamId, this)) {
                sendProcessor.onNext(PayloadFrameCodec.encodeComplete(allocator, streamId));
              }
            }
          }
        };

    sendingSubscriptions.put(streamId, subscriber); 
    response.doOnDiscard(ReferenceCounted.class, DROPPED_ELEMENTS_CONSUMER).subscribe(subscriber);
  }

set isEmpty=false in hookOnNext, but there is no code to set it to true afterwards. sendingSubscriptions.remove(streamId, this) in hookOnComplete has no chance to be executed. cause the subscriber object cannot be recycle.

Why not put sendingSubscriptions.remove(streamId, this) into hookFinally?

[linux-china]

Then link is:

rsocket-java/rsocket-core/src/main/java/io/rsocket/core/RSocketResponder.java

Line 402 in d047c59

private void handleRequestResponse(int streamId, Mono<Payload> response) {

or remove directly in hookOnComplete like logic for requestStream

 protected void hookOnComplete() {
            if (sendingSubscriptions.remove(streamId, this)) {
              sendProcessor.onNext(PayloadFrameCodec.encodeComplete(allocator, streamId));
            }
}

[rstoyanchev]

It does look like sendingSubscriptions.remove should be called in hookOnComplete in any case. Perhaps the condition should be (isEmpty && sendingSubscriptions.remove(streamId, this)). WDYT @OlegDokuka ?

[OlegDokuka]

Looks like a bug. We have to call remove inside hookOnNext for sure.

IsEmpty is an important flag and has to be there in order to prevent Next Complete frame followed by Complete frame