Java 类com.squareup.okhttp.Dispatcher 实例源码

@Test
public void testCustomDispatcher() {
    OkHttpClient client = new OkHttpClient();

    // should be fine with default Dispatcher
    new OkHttpRequestor(client);

    // should also be fine with other common executors that run on separate threads
    client.setDispatcher(new Dispatcher(Executors.newSingleThreadExecutor()));
    new OkHttpRequestor(client);

    client.setDispatcher(new Dispatcher(Executors.newCachedThreadPool()));
    new OkHttpRequestor(client);

    client.setDispatcher(new Dispatcher(Executors.newFixedThreadPool(3)));
    new OkHttpRequestor(client);
}

/**
 * set dispatcher to OkHttpClient
 * @param dispatcher {@link OkHttpClient}.setDispatcher({@link Dispatcher})
 */
public void setDispatcher(Dispatcher dispatcher) {
    if (dispatcher == null) {
        return;
    }
    this.mClient.setDispatcher(dispatcher);
}

@Override public void prepare(final Benchmark benchmark) {
  concurrencyLevel = benchmark.concurrencyLevel;
  targetBacklog = benchmark.targetBacklog;

  client = new OkHttpClient();
  client.setProtocols(benchmark.protocols);
  client.setDispatcher(new Dispatcher(new ThreadPoolExecutor(benchmark.concurrencyLevel,
      benchmark.concurrencyLevel, 60, TimeUnit.SECONDS, new LinkedBlockingQueue<Runnable>())));

  if (benchmark.tls) {
    SSLContext sslContext = SslContextBuilder.localhost();
    SSLSocketFactory socketFactory = sslContext.getSocketFactory();
    HostnameVerifier hostnameVerifier = new HostnameVerifier() {
      @Override public boolean verify(String s, SSLSession session) {
        return true;
      }
    };
    client.setSslSocketFactory(socketFactory);
    client.setHostnameVerifier(hostnameVerifier);
  }

  receiver = new Response.Receiver() {
    @Override public void onFailure(Failure failure) {
      System.out.println("Failed: " + failure.exception());
    }

    @Override public boolean onResponse(Response response) throws IOException {
      Response.Body body = response.body();
      long total = SynchronousHttpClient.readAllAndClose(body.byteStream());
      long finish = System.nanoTime();
      if (VERBOSE) {
        long start = (Long) response.request().tag();
        System.out.printf("Transferred % 8d bytes in %4d ms%n",
            total, TimeUnit.NANOSECONDS.toMillis(finish - start));
      }
      requestsInFlight.decrementAndGet();
      return true;
    }
  };
}

@Test(expectedExceptions={ IllegalArgumentException.class })
public void testSameThreadDispatcher() {
    OkHttpClient client = new OkHttpClient();

    // should fail for same-thread executors
    client.setDispatcher(new Dispatcher(MoreExecutors.newDirectExecutorService()));
    new OkHttpRequestor(client);
}

private ElasticSearchOkHttpClientImpl createClient(MapWrap config) {

        boolean isPrefix;
        String index;
        String url = config.asString("url");
        if (config.exists("index_prefix")) {
            isPrefix = true;
            index = config.asString("index_prefix");
        } else {
            isPrefix = false;
            index = config.asString("index");
        }

        String cacheKey = cacheKey(url, index, isPrefix);
        if (cachedClients.containsKey(cacheKey)) {
            LOGGER.trace("Using cached Elasticsearch client");
            return cachedClients.get(cacheKey);
        }
        LOGGER.trace("Creating new Elasticsearch client");
        final ElasticSearchOkHttpClientImpl es = new ElasticSearchOkHttpClientImpl (
                url,
                index,
                config.asString("type"),
                isPrefix);

        if (config.exists("document_id")) {
            es.withDocumentId(config.asString("document_id"));
        }

        if (config.exists("signer")) {
            es.withSigner(config.getObject("signer"));
        }

        if (config.exists("basic_auth")) {
            if (config.exists("signer")) {
                LOGGER.warn("A client cannot have both signed (AWS) and basic auth. Disabling basic auth");
            } else {
                String auth = config.asString("basic_auth");
                if (!auth.contains(":")) {
                    throw new IllegalArgumentException("Invalid basic_auth value, expected 'user:passwd' but was " + auth);
                }
                if (auth.length() > 1) {
                    String[] split = auth.split(":");
                    es.withBasicAuth(split[0], split[1]);
                }
            }
        }

        if (config.exists("timestamp_field")) {
            es.withTimestampField(config.asString("timestamp_field"));
        }

        if (config.exists("retry")) {
            MapWrap retryConfig = MapWrap.of(config.getObject("retry")).assertExists("policy");
            es.withRetryTimer(Observables.timer(retryConfig), retryConfig.asInt("attempts", DEFAULT_ATTEMPTS));
        }

        if (config.exists("dispatcher")) {
            LOGGER.info("Configuring http dispatcher");
            MapWrap dispatchConfig = MapWrap.of(config.getObject("dispatcher"));
            Dispatcher dispatcher = dispatchConfig.exists("threadpool")
                    ? new Dispatcher(dispatchConfig.getObject("threadpool"))
                    : new Dispatcher();
            dispatcher.setMaxRequests(dispatchConfig.exists("max_concurrent_requests")
                    ? dispatchConfig.asInt("max_concurrent_requests")
                    : dispatcher.getMaxRequests());
            dispatcher.setMaxRequestsPerHost(dispatchConfig.exists("max_concurrent_requests")
                    ? dispatchConfig.asInt("max_concurrent_requests")
                    : dispatcher.getMaxRequestsPerHost());
            es.withDispatcher(dispatcher);
        }

        cachedClients.put(cacheKey, es);

        return es;
    }

public void setDispatcher(Dispatcher dispatcher) {
    if (dispatcher == null) {
        return;
    }
    getDefaultHttpStack().setDispatcher(dispatcher);
}

/**
 * Set an optional {@link Dispatcher} for better control of asynchronous requests.
 *
 * @param dispatcher
 *            the dispatcher
 */
@Inject(optional = true)
public void setDispatcher(final Dispatcher dispatcher) {
    this.dispatcher = dispatcher;
}