java类org.jboss.netty.channel.socket.ClientSocketChannelFactory的实例源码

/**
 * make this factory static thus all clients can share its thread pool.
 * NioClientSocketChannelFactory has only one method newChannel() visible for user, which is thread-safe
 */
public static synchronized ClientSocketChannelFactory getSharedClientChannelFactory(){
  //shared woker and boss pool
  if(factory == null){
    TajoConf conf = new TajoConf();
    int workerNum = conf.getIntVar(TajoConf.ConfVars.INTERNAL_RPC_CLIENT_WORKER_THREAD_NUM);
    factory = createClientChannelFactory("Internal-Client", workerNum);
  }
  return factory;
}

/**
 * Creates a new HttpClientFactory.
 *
 * @param filters the filter chain shared by all Clients created by this factory
 * @param channelFactory the ClientSocketChannelFactory that all Clients created by this
 *          factory will share
 * @param shutdownFactory if true, the channelFactory will be shut down when this
 *          factory is shut down
 * @param executor an executor shared by all Clients created by this factory to schedule
 *          tasks
 * @param shutdownExecutor if true, the executor will be shut down when this factory is
 *          shut down
 */
public HttpClientFactory(FilterChain filters,
                         ClientSocketChannelFactory channelFactory,
                         boolean shutdownFactory,
                         ScheduledExecutorService executor,
                         boolean shutdownExecutor)
{
  this(filters,
       channelFactory,
       shutdownFactory,
       executor,
       shutdownExecutor,
       executor,
       false);
}

/**
 * Creates a new HttpNettyClient
 *
 * @param factory The ClientSocketChannelFactory; it is the caller's responsibility to
 *          shut it down
 * @param executor an executor; it is the caller's responsibility to shut it down
 * @param poolSize Maximum size of the underlying HTTP connection pool
 * @param requestTimeout timeout, in ms, to get a connection from the pool or create one
 * @param idleTimeout interval after which idle connections will be automatically closed
 * @param shutdownTimeout timeout, in ms, the client should wait after shutdown is
 *          initiated before terminating outstanding requests
 * @param maxResponseSize
 * @param sslContext {@link SSLContext}
 * @param sslParameters {@link SSLParameters}with overloaded construct
 * @param queryPostThreshold length of query params above which requests will be tunneled as POSTS
 * @param callbackExecutor an optional executor to invoke user callback
 * @param poolWaiterSize Maximum waiters waiting on the HTTP connection pool
 */
public HttpNettyClient(ClientSocketChannelFactory factory,
                       ScheduledExecutorService executor,
                       int poolSize,
                       int requestTimeout,
                       int idleTimeout,
                       int shutdownTimeout,
                       int maxResponseSize,
                       SSLContext sslContext,
                       SSLParameters sslParameters,
                       int queryPostThreshold,
                       ExecutorService callbackExecutor,
                       int poolWaiterSize)
{
  this(factory,
      executor,
      poolSize,
      requestTimeout,
      idleTimeout,
      shutdownTimeout,
      maxResponseSize,
      sslContext,
      sslParameters,
      queryPostThreshold,
      callbackExecutor,
      poolWaiterSize,
      HttpClientFactory.DEFAULT_CLIENT_NAME,
      HttpClientFactory.NULL_JMX_MANAGER);
}

/**
 * legacy constructor for backward-compatibility purpose.
 */
public HttpNettyClient(ClientSocketChannelFactory factory,
                       ScheduledExecutorService executor,
                       int poolSize,
                       int requestTimeout,
                       int idleTimeout,
                       int shutdownTimeout,
                       int maxResponseSize,
                       SSLContext sslContext,
                       SSLParameters sslParameters,
                       int queryPostThreshold,
                       ExecutorService callbackExecutor,
                       int poolWaiterSize,
                       String name,
                       AbstractJmxManager jmxManager)
{
  this(factory,
      executor,
      poolSize,
      requestTimeout,
      idleTimeout,
      shutdownTimeout,
      maxResponseSize,
      sslContext,
      sslParameters,
      queryPostThreshold,
      callbackExecutor,
      poolWaiterSize,
      name,
      jmxManager,
      AsyncPoolImpl.Strategy.MRU,
      0);
}

/**
 * Creates a new HttpNettyClient
 *
 * @param factory The ClientSocketChannelFactory; it is the caller's responsibility to
 *          shut it down
 * @param executor an executor; it is the caller's responsibility to shut it down
 * @param poolSize Maximum size of the underlying HTTP connection pool
 * @param requestTimeout timeout, in ms, to get a connection from the pool or create one
 * @param idleTimeout interval after which idle connections will be automatically closed
 * @param shutdownTimeout timeout, in ms, the client should wait after shutdown is
 *          initiated before terminating outstanding requests
 * @param maxResponseSize
 * @param sslContext {@link SSLContext}
 * @param sslParameters {@link SSLParameters}with overloaded construct
 * @param queryPostThreshold length of query params above which requests will be tunneled as POSTS
 * @param callbackExecutor an optional executor to invoke user callback
 * @param poolWaiterSize Maximum waiters waiting on the HTTP connection pool
 * @param name Name of the {@link HttpNettyClient}
 * @param jmxManager A management class that is aware of the creation/shutdown event
 *          of the underlying {@link ChannelPoolManager}
 * @param strategy The strategy used to return pool objects.
 * @param minPoolSize Minimum number of objects in the pool. Set to zero for
 *                no minimum.
 */
public HttpNettyClient(ClientSocketChannelFactory factory,
                       ScheduledExecutorService executor,
                       int poolSize,
                       int requestTimeout,
                       int idleTimeout,
                       int shutdownTimeout,
                       int maxResponseSize,
                       SSLContext sslContext,
                       SSLParameters sslParameters,
                       int queryPostThreshold,
                       ExecutorService callbackExecutor,
                       int poolWaiterSize,
                       String name,
                       AbstractJmxManager jmxManager,
                       AsyncPoolImpl.Strategy strategy,
                       int minPoolSize)
{
  _maxResponseSize = maxResponseSize;
  _name = name;
  _channelPoolManager =
      new ChannelPoolManager(new ChannelPoolFactoryImpl(new ClientBootstrap(factory),
          poolSize,
          idleTimeout,
          sslContext,
          sslParameters,
          poolWaiterSize,
          strategy,
          minPoolSize),
          name + ChannelPoolManager.BASE_NAME);
  _scheduler = executor;
  _callbackExecutor = callbackExecutor;
  _requestTimeout = requestTimeout;
  _shutdownTimeout = shutdownTimeout;
  _requestTimeoutMessage = "Exceeded request timeout of " + _requestTimeout + "ms";
  _queryPostThreshold = queryPostThreshold;
  _jmxManager = jmxManager;
  _jmxManager.onProviderCreate(_channelPoolManager);
}

@Test
public void testGetRawClient()
{
  ExecutorService boss = Executors.newCachedThreadPool();
  ExecutorService worker = Executors.newCachedThreadPool();
  ScheduledExecutorService scheduler = Executors.newSingleThreadScheduledExecutor();
  ClientSocketChannelFactory channelFactory = new NioClientSocketChannelFactory(boss, worker);
  HttpClientFactory factory = new HttpClientFactory(FilterChains.empty(), channelFactory, true, scheduler, true);

  Map<String, String> properties = new HashMap<String, String>();

  String requestTimeout = "7000";
  String poolSize = "10";
  String maxResponse = "3000";
  String idleTimeout = "8000";
  String shutdownTimeout = "14000";
  HttpNettyClient client;

  //test creation using default values
  client = factory.getRawClient(properties);
  Assert.assertEquals(client.getMaxResponseSize(), HttpClientFactory.DEFAULT_MAX_RESPONSE_SIZE);
  Assert.assertEquals(client.getRequestTimeout(), HttpClientFactory.DEFAULT_REQUEST_TIMEOUT);
  Assert.assertEquals(client.getShutdownTimeout(), HttpClientFactory.DEFAULT_SHUTDOWN_TIMEOUT);

  //test using only new config keys
  properties.put(HttpClientFactory.HTTP_REQUEST_TIMEOUT, requestTimeout);
  properties.put(HttpClientFactory.HTTP_POOL_SIZE, poolSize);
  properties.put(HttpClientFactory.HTTP_IDLE_TIMEOUT, idleTimeout);
  properties.put(HttpClientFactory.HTTP_MAX_RESPONSE_SIZE, maxResponse);
  properties.put(HttpClientFactory.HTTP_SHUTDOWN_TIMEOUT, shutdownTimeout);
  client = factory.getRawClient(properties);
  Assert.assertEquals(client.getMaxResponseSize(), Integer.parseInt(maxResponse));
  Assert.assertEquals(client.getRequestTimeout(), Integer.parseInt(requestTimeout));
  Assert.assertEquals(client.getShutdownTimeout(), Integer.parseInt(shutdownTimeout));
}

@Test
public void testShutdownTimeout() throws ExecutionException, TimeoutException, InterruptedException
{
  ExecutorService boss = Executors.newCachedThreadPool();
  ExecutorService worker = Executors.newCachedThreadPool();
  ScheduledExecutorService scheduler = Executors.newSingleThreadScheduledExecutor();
  ClientSocketChannelFactory channelFactory = new NioClientSocketChannelFactory(boss, worker);
  HttpClientFactory factory = new HttpClientFactory(FilterChains.empty(), channelFactory, true, scheduler, true);

  List<Client> clients = new ArrayList<Client>();
  for (int i = 0; i < 100; i++)
  {
    clients.add(new TransportClientAdapter(factory.getClient(Collections.<String, String>emptyMap())));
  }

  for (Client c : clients)
  {
    RestRequest r = new RestRequestBuilder(_testServer.getRequestURI()).build();
    c.restRequest(r).get(30, TimeUnit.SECONDS);
  }

  FutureCallback<None> factoryShutdown = new FutureCallback<None>();
  factory.shutdown(factoryShutdown, 1, TimeUnit.SECONDS);

  factoryShutdown.get(30, TimeUnit.SECONDS);

  Assert.assertTrue(boss.awaitTermination(30, TimeUnit.SECONDS), "Failed to shut down boss");
  Assert.assertTrue(worker.awaitTermination(30, TimeUnit.SECONDS), "Failed to shut down worker");
  Assert.assertTrue(scheduler.awaitTermination(30, TimeUnit.SECONDS), "Failed to shut down scheduler");
}

@Test
public void testShutdownNoTimeout() throws ExecutionException, TimeoutException, InterruptedException
{
  ExecutorService boss = Executors.newCachedThreadPool();
  ExecutorService worker = Executors.newCachedThreadPool();
  ScheduledExecutorService scheduler = Executors.newSingleThreadScheduledExecutor();
  ClientSocketChannelFactory channelFactory = new NioClientSocketChannelFactory(boss, worker);
  HttpClientFactory factory = new HttpClientFactory(FilterChains.empty(), channelFactory, true, scheduler, true);

  List<Client> clients = new ArrayList<Client>();
  for (int i = 0; i < 100; i++)
  {
    clients.add(new TransportClientAdapter(factory.getClient(Collections.<String, String>emptyMap())));
  }

  for (Client c : clients)
  {
    RestRequest r = new RestRequestBuilder(_testServer.getRequestURI()).build();
    c.restRequest(r).get(30, TimeUnit.SECONDS);
  }

  FutureCallback<None> factoryShutdown = new FutureCallback<None>();
  factory.shutdown(factoryShutdown);

  try
  {
    factoryShutdown.get(1, TimeUnit.SECONDS);
    Assert.fail("Factory shutdown should have timed out");
  }
  catch (TimeoutException e)
  {
    // Expected
  }

  Assert.assertFalse(boss.isShutdown(), "Boss should not be shut down");
  Assert.assertFalse(worker.isShutdown(), "Worker should not be shut down");
  Assert.assertFalse(scheduler.isShutdown(), "Scheduler should not be shut down");
}

@Test
public void testShutdownIOThread() throws ExecutionException, TimeoutException, InterruptedException
{
  ExecutorService boss = Executors.newCachedThreadPool();
  ExecutorService worker = Executors.newCachedThreadPool();
  ScheduledExecutorService scheduler = Executors.newSingleThreadScheduledExecutor();
  ClientSocketChannelFactory channelFactory = new NioClientSocketChannelFactory(boss, worker);
  HttpClientFactory factory = new HttpClientFactory(FilterChains.empty(), channelFactory, true, scheduler, true);

  Client client = new TransportClientAdapter(factory.getClient(
          Collections.<String, Object>emptyMap()));

  Future<RestResponse> responseFuture = client.restRequest(new RestRequestBuilder(_testServer.resetResponseLatch(1)).build());


  FutureCallback<None> factoryShutdown = new FutureCallback<None>();
  factory.shutdown(factoryShutdown);

  FutureCallback<None> clientShutdown = new FutureCallback<None>();
  client.shutdown(clientShutdown);

  // Client and factory shutdowns are now pending.  When we release the latch, the response will
  // be returned, which causes the shutdowns to complete on the Netty IO thread that received the
  // response.
  _testServer.releaseResponseLatch();

  responseFuture.get(60, TimeUnit.SECONDS);
  clientShutdown.get(60, TimeUnit.SECONDS);
  factoryShutdown.get(60, TimeUnit.SECONDS);

  Assert.assertTrue(boss.awaitTermination(60, TimeUnit.SECONDS));
  Assert.assertTrue(worker.awaitTermination(60, TimeUnit.SECONDS));
  Assert.assertTrue(scheduler.awaitTermination(60, TimeUnit.SECONDS));
}

public PerChannelBookieClient(OrderedSafeExecutor executor, ClientSocketChannelFactory channelFactory,
        InetSocketAddress addr, AtomicLong totalBytesOutstanding) {
    this.addr = addr;
    this.executor = executor;
    this.totalBytesOutstanding = totalBytesOutstanding;
    this.channelFactory = channelFactory;
    connect(channelFactory);
}