docs: public Flux<ServiceMessage> requestBidirectional(
        Publisher<ServiceMessage> publisher, Class<?> responseType) {
      return Flux.from(publisher)
          .switchOnFirst(
              (first, messages) -> {
                if (first.hasValue()) {
                  ServiceMessage request = first.get();
                  String qualifier = request.qualifier();
  
                  if (methodRegistry.containsInvoker(qualifier)) { // local service.
                    return methodRegistry
                        .getInvoker(qualifier)
                        .invokeBidirectional(messages, ServiceMessageCodec::decodeData)
                        .map(this::throwIfError);
                  } else {
                    // remote service
                    return addressLookup(request)
                        .flatMapMany(
                            address -> requestBidirectional(messages, responseType, address));
                  }
                }
  
                return messages;
              });
    }