docs: public static Function<RSocket, RSocket> safeClose() {
      return source ->
          new RSocketProxy(source) {
            final AtomicInteger count = new AtomicInteger();
            final AtomicBoolean closed = new AtomicBoolean();
  
            @Override
            public Mono<Void> fireAndForget(Payload payload) {
              return source
                  .fireAndForget(payload)
                  .doOnSubscribe(s -> count.incrementAndGet())
                  .doFinally(
                      signalType -> {
                        if (count.decrementAndGet() == 0 && closed.get()) {
                          source.dispose();
                        }
                      });
            }
  
            @Override
            public Mono<Payload> requestResponse(Payload payload) {
              return source
                  .requestResponse(payload)
                  .doOnSubscribe(s -> count.incrementAndGet())
                  .doFinally(
                      signalType -> {
                        if (count.decrementAndGet() == 0 && closed.get()) {
                          source.dispose();
                        }
                      });
            }
  
            @Override
            public Flux<Payload> requestStream(Payload payload) {
              return source
                  .requestStream(payload)
                  .doOnSubscribe(s -> count.incrementAndGet())
                  .doFinally(
                      signalType -> {
                        if (count.decrementAndGet() == 0 && closed.get()) {
                          source.dispose();
                        }
                      });
            }
  
            @Override
            public Flux<Payload> requestChannel(Publisher<Payload> payloads) {
              return source
                  .requestChannel(payloads)
                  .doOnSubscribe(s -> count.incrementAndGet())
                  .doFinally(
                      signalType -> {
                        if (count.decrementAndGet() == 0 && closed.get()) {
                          source.dispose();
                        }
                      });
            }
  
            @Override
            public Mono<Void> metadataPush(Payload payload) {
              return source
                  .metadataPush(payload)
                  .doOnSubscribe(s -> count.incrementAndGet())
                  .doFinally(
                      signalType -> {
                        if (count.decrementAndGet() == 0 && closed.get()) {
                          source.dispose();
                        }
                      });
            }
  
            @Override
            public void dispose() {
              if (closed.compareAndSet(false, true)) {
                if (count.get() == 0) {
                  source.dispose();
                }
              }
            }
          };
    }