throw exception on errorConsumer, client will not finish #1105

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);
}

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();
}