throw exception on errorConsumer, client will not finish

[levil3]

I tried throwing an exception in the errorConsumer on the server, so that the client's request seems to not be released, and will not do doOnComplete() and map()

Expected Behavior

I expect it to print 10 times "complete 1"

Actual Behavior

But it didn't print even once

Steps to Reproduce

my code is :

• server

  public static void main(String[] args) throws InterruptedException {
      RSocket rsocket = new RSocket() {
          @Override
          public Flux<Payload> requestStream(Payload p) {
              return Flux.create(emitter -> Mono.just("test").subscribe(payload -> {
                  throw new RuntimeException();
              }, throwable -> {
                  throw new RuntimeException();
              }, emitter::complete));
          }
      };
  
      LoopResources resources = LoopResources.create("test", 1, 2, true);
      TcpServer server = TcpServer.create().host("localhost").port(8090).runOn(resources);
      RSocketServer.create(SocketAcceptor.with(rsocket))
              .bind(TcpServerTransport.create(server))
              .subscribe();
  
      // wait
      TimeUnit.MINUTES.sleep(10);
  }

• client

  public static void main(String[] args) throws InterruptedException {
      RSocket rSocket = RSocketConnector.create()
              .payloadDecoder(PayloadDecoder.ZERO_COPY)
              .connect(TcpClientTransport.create(8090))
              .doOnSuccess(rSocket1 -> {
                  System.out.println("connect server success!");
              })
              .block();
  
      for (int i = 0; i < 10; ++i) {
          Payload payload = ByteBufPayload.create("0");
          rSocket.requestStream(payload).map(Payload::getDataUtf8).doOnComplete(() -> {
              System.out.println("complete 1");
          }).map(e -> {
              System.out.println("doMap");
              return e;
          }).blockLast();
      }
  
      Thread.sleep(TimeUnit.SECONDS.toNanos(1000));
  
      // dispose
      rSocket.dispose();
  }

Possible Solution

Your Environment

• RSocket version(s) used: v1.1.4
• jdk: 1.8