import com.netflix.loadbalancer.reactive.LoadBalancerCommand; //导入依赖的package包/类
@Override
public AsyncFuture<Void> sendBatches(final Collection<Batch> batches) {
    final List<Callable<AsyncFuture<Void>>> callables = new ArrayList<>();

    for (final Batch b : batches) {
        callables.add(() -> {
            final Observable<Void> observable = LoadBalancerCommand.<Void>builder()
                .withLoadBalancer(loadBalancer)
                .build()
                .submit(server -> toObservable(clientFactory.newClient(server).sendBatch(b)));

            return fromObservable(observable);
        });
    }

    return async.eventuallyCollect(callables, VOID_COLLECTOR, PARALLELISM);
}
 

import com.netflix.loadbalancer.reactive.LoadBalancerCommand; //导入依赖的package包/类
protected LoadBalancingHttpClient(Builder<I, O> builder) {
    super(builder.lb, builder.config, new RequestSpecificRetryHandler(true, true, builder.retryHandler, null), builder.pipelineConfigurator, builder.poolCleanerScheduler);
    requestIdHeaderName = getProperty(IClientConfigKey.Keys.RequestIdHeaderName, null, null);
    requestIdProvider = (requestIdHeaderName != null) 
                      ? new HttpRequestIdProvider(requestIdHeaderName, RxContexts.DEFAULT_CORRELATOR)
                      : null;
    this.listeners = new CopyOnWriteArrayList<ExecutionListener<HttpClientRequest<I>, HttpClientResponse<O>>>(builder.listeners);
    defaultCommandBuilder = LoadBalancerCommand.<HttpClientResponse<O>>builder()
            .withLoadBalancerContext(lbContext)
            .withListeners(this.listeners)
            .withClientConfig(builder.config)
            .withRetryHandler(builder.retryHandler)
            .build();
    this.responseToErrorPolicy = builder.responseToErrorPolicy;
    this.backoffStrategy = builder.backoffStrategy;
}
 

import com.netflix.loadbalancer.reactive.LoadBalancerCommand; //导入依赖的package包/类
public Observable<DatagramPacket> submit(final String content) {
    return LoadBalancerCommand.<DatagramPacket>builder()
            .withLoadBalancerContext(lbContext)
            .build()
            .submit(new ServerOperation<DatagramPacket>() {
                @Override
                public Observable<DatagramPacket> call(Server server) {
                    RxClient<DatagramPacket, DatagramPacket> rxClient = getOrCreateRxClient(server);
                    return rxClient.connect().flatMap(new Func1<ObservableConnection<DatagramPacket, DatagramPacket>, Observable<? extends DatagramPacket>>() {
                        @Override
                        public Observable<? extends DatagramPacket> call(ObservableConnection<DatagramPacket, DatagramPacket> connection) {
                            connection.writeStringAndFlush(content);
                            return connection.getInput().timeout(10, TimeUnit.MILLISECONDS).take(1);
                        }
                    });
                }
            });
}
 

import com.netflix.loadbalancer.reactive.LoadBalancerCommand; //导入依赖的package包/类
public String call(final String path) throws Exception {
    return LoadBalancerCommand.<String>builder()
            .withLoadBalancer(loadBalancer)
            .build()
            .submit(new ServerOperation<String>() {
        @Override
        public Observable<String> call(Server server) {
            URL url;
            try {
                url = new URL("http://" + server.getHost() + ":" + server.getPort() + path);
                HttpURLConnection conn = (HttpURLConnection) url.openConnection();
                return Observable.just(conn.getResponseMessage());
            } catch (Exception e) {
                return Observable.error(e);
            }
        }
    }).toBlocking().first();
}
 

import com.netflix.loadbalancer.reactive.LoadBalancerCommand; //导入依赖的package包/类
private <T, R> R loadBalanceRESTOperation(String endpointURI,
   		Class<T> entityClass, Function<RESTClient<T>, R> operation)
   				throws NotFoundException, LoadBalancerTimeoutException {
   	R r = null;
   	loadBalancerModificationLock.readLock().lock();
   	try {
   		if (loadBalancer == null) {
       		LOG.warn("Load Balancer was not initialized for service: " + targetService.getServiceName()
       			+ ". Is Registry up?");
       		updateLoadBalancersForServiceUsingRegistry(targetService);
       	}
       	if (loadBalancer == null || loadBalancer.getAllServers().isEmpty()) {
       		LOG.warn("No Server registered for Service: " + targetService.getServiceName());
       	} else {
       		ServiceLoadBalancerResult<R> slbr = LoadBalancerCommand.<ServiceLoadBalancerResult<R>>builder()
                       .withLoadBalancer(loadBalancer)
                       .withRetryHandler(retryHandler)
                       .build()
                       .submit(server -> Observable.just(
                       		ServiceLoadBalancerResult.fromRESTOperation(
                       				(RESTClient<T>) getEndpointClientCollection(endpointURI, entityClass)
               				.getRESTClient(server), operation)
                       		))
                       .onErrorReturn((Throwable e) -> null).toBlocking().first();
       		if (slbr == null || slbr.getStatusCode() == Status.REQUEST_TIMEOUT.getStatusCode()) {
       			throw new LoadBalancerTimeoutException("Timout at endpoint: "
       					+ endpointURI + ", with target service: " + targetService.getServiceName(),
       					targetService);
       		} else if (slbr.getStatusCode() == Status.NOT_FOUND.getStatusCode() || slbr.getEntity() == null) {
       			throw new NotFoundException();
       		}
       		r = slbr.getEntity();
       	}
   	} finally {
   		loadBalancerModificationLock.readLock().unlock();
   	}
	return r;
}
 

import com.netflix.loadbalancer.reactive.LoadBalancerCommand; //导入依赖的package包/类
public Observable<ResponseEntity<Message>> rxFilter(Message input) {
    return LoadBalancerCommand.<ResponseEntity<Message>>builder()
            .withLoadBalancer(loadBalancer)
            .withRetryHandler(retryHandler)
            .build()
            .submit(server -> Observable.just(callRemoteService(input, server)));
}
 

import com.netflix.loadbalancer.reactive.LoadBalancerCommand; //导入依赖的package包/类
public Observable<ResponseEntity<Version>> rxFilterVersion() {
    return LoadBalancerCommand.<ResponseEntity<Version>>builder()
            .withLoadBalancer(loadBalancer)
            .withRetryHandler(retryHandler)
            .build()
            .submit(server -> Observable.just(getFilterServiceVersion(server)));
}
 

import com.netflix.loadbalancer.reactive.LoadBalancerCommand; //导入依赖的package包/类
public Observable<ResponseEntity<Message>> rxStore(Message input) {
    return LoadBalancerCommand
            .<ResponseEntity<Message>>builder()
            .withLoadBalancer(loadBalancer)
            .withRetryHandler(retryHandler)
            .build()
            .submit(
                    server -> Observable.just(callRemoteStoreService(input, server))
            );
}
 

import com.netflix.loadbalancer.reactive.LoadBalancerCommand; //导入依赖的package包/类
public Observable<ResponseEntity<List<Message>>> rxGet() {
    return LoadBalancerCommand
            .<ResponseEntity<List<Message>>>builder()
            .withLoadBalancer(loadBalancer)
            .withRetryHandler(retryHandler)
            .build()
            .submit(
                    server -> Observable.just(callRemoteGetService(server))
            );
}
 

import com.netflix.loadbalancer.reactive.LoadBalancerCommand; //导入依赖的package包/类
public Observable<ResponseEntity<Version>> rxStorageVersion() {
    return LoadBalancerCommand.<ResponseEntity<Version>>builder()
            .withLoadBalancer(loadBalancer)
            .withRetryHandler(retryHandler)
            .build()
            .submit(server -> Observable.just(getStorageServiceVersion(server)));
}
 

import com.netflix.loadbalancer.reactive.LoadBalancerCommand; //导入依赖的package包/类
@Autowired
public ProxyController(HttpProxy httpProxy, SpringClientFactory springClientFactory) {
    this.httpProxy = httpProxy;

    RibbonLoadBalancerContext context = springClientFactory.getLoadBalancerContext(SERVICEID);
    IClientConfig clientConfig = springClientFactory.getClientConfig(SERVICEID);
    ILoadBalancer loadBalancer = springClientFactory.getLoadBalancer(SERVICEID);
    HttpClientLoadBalancerErrorHandler requestSpecificRetryHandler = getRequestSpecificRetryHandler(clientConfig);

    this.commandBuilder = LoadBalancerCommand.builder()
            .withRetryHandler(requestSpecificRetryHandler)
            .withLoadBalancerContext(context)
            .withClientConfig(clientConfig)
            .withLoadBalancer(loadBalancer);
}
 

import com.netflix.loadbalancer.reactive.LoadBalancerCommand; //导入依赖的package包/类
protected LoadBalancerCommand<T> buildLoadBalancerCommand(final S request, final IClientConfig config) {
	RequestSpecificRetryHandler handler = getRequestSpecificRetryHandler(request, config);
	LoadBalancerCommand.Builder<T> builder = LoadBalancerCommand.<T>builder()
			.withLoadBalancerContext(this)
			.withRetryHandler(handler)
			.withLoadBalancerURI(request.getUri());
	customizeLoadBalancerCommandBuilder(request, config, builder);
	return builder.build();
}
 

import com.netflix.loadbalancer.reactive.LoadBalancerCommand; //导入依赖的package包/类
/**
 * Subject an operation to run in the load balancer
 * 
 * @param request
 * @param errorHandler
 * @param requestConfig
 * @param rxClientConfig
 * @return
 */
private Observable<HttpClientResponse<O>> submit(final Server server, final HttpClientRequest<I> request, final RetryHandler errorHandler, final IClientConfig requestConfig, final ClientConfig rxClientConfig) {
    RetryHandler retryHandler = errorHandler;
    if (retryHandler == null) {
        retryHandler = getRequestRetryHandler(request, requestConfig);
    }
    
    final IClientConfig config = requestConfig == null ? DefaultClientConfigImpl.getEmptyConfig() : requestConfig;
    final ExecutionContext<HttpClientRequest<I>> context = new ExecutionContext<HttpClientRequest<I>>(request, config, this.getClientConfig(), retryHandler);
    
    Observable<HttpClientResponse<O>> result = submitToServerInURI(request, config, rxClientConfig, retryHandler, context);
    if (result == null) {
        LoadBalancerCommand<HttpClientResponse<O>> command;
        if (retryHandler != defaultRetryHandler) {
            // need to create new builder instead of the default one
            command = LoadBalancerCommand.<HttpClientResponse<O>>builder()
                    .withExecutionContext(context)
                    .withLoadBalancerContext(lbContext)
                    .withListeners(listeners)
                    .withClientConfig(this.getClientConfig())
                    .withRetryHandler(retryHandler)
                    .withServer(server)
                    .build();
        }
        else {
            command = defaultCommandBuilder;
        }
        
        result = command.submit(requestToOperation(request, getRxClientConfig(config, rxClientConfig)));
    }
    return result;
}
 

import com.netflix.loadbalancer.reactive.LoadBalancerCommand; //导入依赖的package包/类
/**
 * Submits the request to the server indicated in the URI
 * @param request
 * @param requestConfig
 * @param config
 * @param errorHandler
 * @param context
 * @return
 */
private Observable<HttpClientResponse<O>> submitToServerInURI(
        HttpClientRequest<I> request, IClientConfig requestConfig, ClientConfig config,
        RetryHandler errorHandler, ExecutionContext<HttpClientRequest<I>> context)  {
    // First, determine server from the URI
    URI uri;
    try {
        uri = new URI(request.getUri());
    } catch (URISyntaxException e) {
        return Observable.error(e);
    }
    String host = uri.getHost();
    if (host == null) {
        return null;
    }
    int port = uri.getPort();
    if (port < 0) {
        if (clientConfig.getPropertyAsBoolean(IClientConfigKey.Keys.IsSecure, false)) {
            port = 443;
        } else {
            port = 80;
        }
    }
    
    return LoadBalancerCommand.<HttpClientResponse<O>>builder()
            .withRetryHandler(errorHandler)
            .withLoadBalancerContext(lbContext)
            .withListeners(listeners)
            .withExecutionContext(context)
            .withServer(new Server(host, port))
            .build()
            .submit(this.requestToOperation(request, getRxClientConfig(requestConfig, config)));
}
 

import com.netflix.loadbalancer.reactive.LoadBalancerCommand; //导入依赖的package包/类
@Override
public Observable<ObservableConnection<O, I>> connect() {
    return LoadBalancerCommand.<ObservableConnection<O, I>>builder()
            .withLoadBalancerContext(lbContext)
            .build()
            .submit(new ServerOperation<ObservableConnection<O, I>>() {
                @Override
                public Observable<ObservableConnection<O, I>> call(Server server) {
                    return getOrCreateRxClient(server).connect();            
                }                    
            });
}
 

import com.netflix.loadbalancer.reactive.LoadBalancerCommand; //导入依赖的package包/类
protected LoadBalancerCommand<Void> buildCommand() {
    return LoadBalancerCommand.<Void>builder().withLoadBalancer(loadBalancer).build();
}
 

import com.netflix.loadbalancer.reactive.LoadBalancerCommand; //导入依赖的package包/类
@Override
protected void customizeLoadBalancerCommandBuilder(S request, IClientConfig config, LoadBalancerCommand.Builder<T> builder) {
	if (request.getLoadBalancerKey() != null) {
		builder.withServerLocator(request.getLoadBalancerKey());
	}
}
 

import com.netflix.loadbalancer.reactive.LoadBalancerCommand; //导入依赖的package包/类
protected void customizeLoadBalancerCommandBuilder(final S request, final IClientConfig config,
		final LoadBalancerCommand.Builder<T> builder) {
	// do nothing by default, give a chance to its derived class to customize the builder
}