code: public Flux<ServiceMessage> requestMany(ServiceMessage request, Class<?> responseType) {
      return Flux.defer(
          () -> {
            String qualifier = request.qualifier();
            if (methodRegistry.containsInvoker(qualifier)) { // local service.
              return methodRegistry
                  .getInvoker(request.qualifier())
                  .invokeMany(request, ServiceMessageCodec::decodeData)
                  .map(this::throwIfError);
            } else {
              return addressLookup(request)
                  .flatMapMany(
                      address -> requestMany(request, responseType, address)); // remote service
            }
          });
    }
docs: Issues request to service which returns stream of service messages back.
  
  @param request request with given headers.
  @param responseType type of responses.
  @return flux publisher of service responses.