Java 类org.apache.thrift.transport.TTransportFactory 实例源码

@Bean(name = "pool-server")
public TServer poolServer() throws Exception {
    TServerTransport transport = new TServerSocket(this.port());

    TThreadPoolServer.Args args = new TThreadPoolServer.Args(transport);
    args.transportFactory(new TTransportFactory());
    args.protocolFactory(new TBinaryProtocol.Factory());

    args.processor(this.processor());
    args.executorService(new ThreadPoolExecutor(env.getProperty(
            "rpc.server.min.worker.threads", Integer.class, 512), env
            .getProperty("rpc.server.max.worker.threads", Integer.class,
                    65535), env.getProperty(
            "rpc.server.thread.keep.alive.time", Long.class, 600l),
            TimeUnit.SECONDS, new SynchronousQueue<Runnable>()));

    return new TThreadPoolServer(args);
}

public ApacheThriftMethodInvoker(
        ListeningExecutorService executorService,
        ListeningScheduledExecutorService delayService,
        TTransportFactory transportFactory,
        TProtocolFactory protocolFactory,
        Duration connectTimeout,
        Duration requestTimeout,
        Optional<HostAndPort> socksProxy,
        Optional<SSLContext> sslContext)
{
    this.executorService = requireNonNull(executorService, "executorService is null");
    this.delayService = requireNonNull(delayService, "delayService is null");
    this.transportFactory = requireNonNull(transportFactory, "transportFactory is null");
    this.protocolFactory = requireNonNull(protocolFactory, "protocolFactory is null");
    this.connectTimeoutMillis = Ints.saturatedCast(requireNonNull(connectTimeout, "connectTimeout is null").toMillis());
    this.requestTimeoutMillis = Ints.saturatedCast(requireNonNull(requestTimeout, "requestTimeout is null").toMillis());
    this.socksProxy = requireNonNull(socksProxy, "socksProxy is null");
    this.sslContext = requireNonNull(sslContext, "sslContext is null");
}

private TTransportFactory getSASLTransportFactory() {
  String[] names;
  try {
    names = FlumeAuthenticationUtil.splitKerberosName(principal);
  } catch (IOException e) {
    throw new FlumeException(
            "Error while trying to resolve Principal name - " + principal, e);
  }
  Map<String, String> saslProperties = new HashMap<String, String>();
  saslProperties.put(Sasl.QOP, "auth");
  TSaslServerTransport.Factory saslTransportFactory =
          new TSaslServerTransport.Factory();
  saslTransportFactory.addServerDefinition(
          "GSSAPI", names[0], names[1], saslProperties,
          FlumeAuthenticationUtil.getSaslGssCallbackHandler());
  return saslTransportFactory;
}

private static TServer getTThreadPoolServer(TProtocolFactory protocolFactory,
                                            TProcessor processor,
                                            TTransportFactory transportFactory,
                                            int workerThreads,
                                            InetSocketAddress inetSocketAddress,
                                            int backlog,
                                            int clientTimeout)
    throws TTransportException {
  TServerTransport serverTransport = new TServerSocket(
                                         new TServerSocket.ServerSocketTransportArgs().
                                             bindAddr(inetSocketAddress).backlog(backlog).
                                             clientTimeout(clientTimeout));
  log.info("starting HBase ThreadPool Thrift server on " + inetSocketAddress.toString());
  TThreadPoolServer.Args serverArgs = new TThreadPoolServer.Args(serverTransport);
  serverArgs.processor(processor);
  serverArgs.transportFactory(transportFactory);
  serverArgs.protocolFactory(protocolFactory);
  if (workerThreads > 0) {
    serverArgs.maxWorkerThreads(workerThreads);
  }
  return new TThreadPoolServer(serverArgs);
}

public void start(CountDownLatch latch, int port) {
    try {
        TNonblockingServerSocket serverTransport = new TNonblockingServerSocket(port);
        //异步IO，需要使用TFramedTransport，它将分块缓存读取。
        TTransportFactory transportFactory = new TFramedTransport.Factory();
        //使用高密度二进制协议
        TProtocolFactory proFactory = new TBinaryProtocol.Factory();
        //发布多个服务
        TMultiplexedProcessor processor = new TMultiplexedProcessor();
        processor.registerProcessor(ClassNameUtils.getClassName(Hello.class), new Hello.Processor<>(new HelloServer()));

        TServer server = new TThreadedSelectorServer(new
                TThreadedSelectorServer.Args(serverTransport)
                .transportFactory(transportFactory)
                .protocolFactory(proFactory)
                .processor(processor)
        );
        System.out.println("Starting the hello server...");
        latch.countDown();
        server.serve();
    } catch (Exception e) {
        e.printStackTrace();
    }
}

/**
 * Start the SessionDriver callback server
 * @param conf the corona configuration for this session
 * @return the server socket of the callback server
 * @throws IOException
 */
private ServerSocket initializeServer(CoronaConf conf) throws IOException {
  // Choose any free port.
  ServerSocket sessionServerSocket =
    new ServerSocket(0, 0, getLocalAddress());
  TServerSocket tServerSocket = new TServerSocket(sessionServerSocket,
                                                  conf.getCMSoTimeout());

  TFactoryBasedThreadPoolServer.Args args =
    new TFactoryBasedThreadPoolServer.Args(tServerSocket);
  args.processor(new SessionDriverServiceProcessor(incoming));
  args.transportFactory(new TTransportFactory());
  args.protocolFactory(new TBinaryProtocol.Factory(true, true));
  args.stopTimeoutVal = 0;
  server = new TFactoryBasedThreadPoolServer(
    args, new TFactoryBasedThreadPoolServer.DaemonThreadFactory());
  return sessionServerSocket;
}

/**
 * Constrcts a server object
 */
public HadoopThriftServer(String [] args) {

  if (args.length > 0) {
    serverPort = new Integer(args[0]);
  }
  try {
    ServerSocket ssock = createServerSocket(serverPort);
    TServerTransport serverTransport = new TServerSocket(ssock);
    Iface handler = new HadoopThriftHandler("hdfs-thrift-dhruba");
    ThriftHadoopFileSystem.Processor processor = new ThriftHadoopFileSystem.Processor(handler);
    TThreadPoolServer.Args serverArgs = new TThreadPoolServer.Args(serverTransport);
    serverArgs.minWorkerThreads = 10;
    serverArgs.processor(processor);
    serverArgs.transportFactory(new TTransportFactory());
    serverArgs.protocolFactory(new TBinaryProtocol.Factory());
    server = new TThreadPoolServer(serverArgs);
    System.out.println("Starting the hadoop thrift server on port [" + serverPort + "]...");
    HadoopThriftHandler.LOG.info("Starting the hadoop thrift server on port [" +serverPort + "]...");
    System.out.flush();

  } catch (Exception x) {
    x.printStackTrace();
  }
}

public ThriftServerThread(InetAddress listenAddr,
                          int listenPort,
                          int listenBacklog,
                          TProcessor processor,
                          TTransportFactory transportFactory)
{
    // now we start listening for clients
    logger.info(String.format("Binding thrift service to %s:%s", listenAddr, listenPort));

    TServerFactory.Args args = new TServerFactory.Args();
    args.tProtocolFactory = new TBinaryProtocol.Factory(true, true);
    args.addr = new InetSocketAddress(listenAddr, listenPort);
    args.listenBacklog = listenBacklog;
    args.processor = processor;
    args.keepAlive = DatabaseDescriptor.getRpcKeepAlive();
    args.sendBufferSize = DatabaseDescriptor.getRpcSendBufferSize();
    args.recvBufferSize = DatabaseDescriptor.getRpcRecvBufferSize();
    args.inTransportFactory = transportFactory;
    args.outTransportFactory = transportFactory;
    serverEngine = new TServerCustomFactory(DatabaseDescriptor.getRpcServerType()).buildTServer(args);
}

public ThriftServerThread(InetAddress listenAddr,
                          int listenPort,
                          CassandraServer server,
                          TProcessor processor,
                          TTransportFactory transportFactory)
{
    // now we start listening for clients
    logger.info(String.format("Binding thrift service to %s:%s", listenAddr, listenPort));

    TServerFactory.Args args = new TServerFactory.Args();
    args.tProtocolFactory = new TBinaryProtocol.Factory(true, true);
    args.addr = new InetSocketAddress(listenAddr, listenPort);
    args.cassandraServer = server;
    args.processor = processor;
    args.keepAlive = DatabaseDescriptor.getRpcKeepAlive();
    args.sendBufferSize = DatabaseDescriptor.getRpcSendBufferSize();
    args.recvBufferSize = DatabaseDescriptor.getRpcRecvBufferSize();
    args.inTransportFactory = transportFactory;
    args.outTransportFactory = transportFactory;
    serverEngine = new TServerCustomFactory(DatabaseDescriptor.getRpcServerType()).buildTServer(args);
}

private TTransportFactory validateAndSetTransportFactory(String transportFactory)
{
    try
    {
        Class factory = Class.forName(transportFactory);

        if(!TTransportFactory.class.isAssignableFrom(factory))
            throw new IllegalArgumentException(String.format("transport factory '%s' " +
                    "not derived from TTransportFactory", transportFactory));

        return (TTransportFactory) factory.newInstance();
    }
    catch (Exception e)
    {
        throw new IllegalArgumentException(String.format("Cannot create a transport factory '%s'.", transportFactory), e);
    }
}

@Override
public TServer getServer(TProcessor processor) throws IOException, TTransportException {
    int port = type.getPort(storm_conf);
    TTransportFactory serverTransportFactory = getServerTransportFactory();
    TServerSocket serverTransport = new TServerSocket(port);
    int numWorkerThreads = type.getNumThreads(storm_conf);
    Integer queueSize = type.getQueueSize(storm_conf);

    TThreadPoolServer.Args server_args =
            new TThreadPoolServer.Args(serverTransport).processor(new TUGIWrapProcessor(processor)).minWorkerThreads(numWorkerThreads)
                    .maxWorkerThreads(numWorkerThreads).protocolFactory(new TBinaryProtocol.Factory(false, true));

    if (serverTransportFactory != null) {
        server_args.transportFactory(serverTransportFactory);
    }
    BlockingQueue workQueue = new SynchronousQueue();
    if (queueSize != null) {
        workQueue = new ArrayBlockingQueue(queueSize);
    }
    ThreadPoolExecutor executorService = new ExtendedThreadPoolExecutor(numWorkerThreads, numWorkerThreads, 60, TimeUnit.SECONDS, workQueue);
    server_args.executorService(executorService);
    return new TThreadPoolServer(server_args);
}

public ThriftServerThread(InetAddress listenAddr,
                          int listenPort,
                          int listenBacklog,
                          TProcessor processor,
                          TTransportFactory transportFactory)
{
    // now we start listening for clients
    logger.info(String.format("Binding thrift service to %s:%s", listenAddr, listenPort));

    TServerFactory.Args args = new TServerFactory.Args();
    args.tProtocolFactory = new TBinaryProtocol.Factory(true, true);
    args.addr = new InetSocketAddress(listenAddr, listenPort);
    args.listenBacklog = listenBacklog;
    args.processor = processor;
    args.keepAlive = DatabaseDescriptor.getRpcKeepAlive();
    args.sendBufferSize = DatabaseDescriptor.getRpcSendBufferSize();
    args.recvBufferSize = DatabaseDescriptor.getRpcRecvBufferSize();
    args.inTransportFactory = transportFactory;
    args.outTransportFactory = transportFactory;
    serverEngine = new TServerCustomFactory(DatabaseDescriptor.getRpcServerType()).buildTServer(args);
}

public ThriftServerThread(InetAddress listenAddr,
                          int listenPort,
                          int listenBacklog,
                          TProcessor processor,
                          TTransportFactory transportFactory)
{
    // now we start listening for clients
    logger.info(String.format("Binding thrift service to %s:%s", listenAddr, listenPort));

    TServerFactory.Args args = new TServerFactory.Args();
    args.tProtocolFactory = new TBinaryProtocol.Factory(true, true);
    args.addr = new InetSocketAddress(listenAddr, listenPort);
    args.listenBacklog = listenBacklog;
    args.processor = processor;
    args.keepAlive = DatabaseDescriptor.getRpcKeepAlive();
    args.sendBufferSize = DatabaseDescriptor.getRpcSendBufferSize();
    args.recvBufferSize = DatabaseDescriptor.getRpcRecvBufferSize();
    args.inTransportFactory = transportFactory;
    args.outTransportFactory = transportFactory;
    serverEngine = new TServerCustomFactory(DatabaseDescriptor.getRpcServerType()).buildTServer(args);
}

private static TServer getTHsHaServer(TProtocolFactory protocolFactory,
    TProcessor processor, TTransportFactory transportFactory,
    int workerThreads, int maxCallQueueSize,
    InetSocketAddress inetSocketAddress, ThriftMetrics metrics)
    throws TTransportException {
  TNonblockingServerTransport serverTransport = new TNonblockingServerSocket(inetSocketAddress);
  log.info("starting HBase HsHA Thrift server on " + inetSocketAddress.toString());
  THsHaServer.Args serverArgs = new THsHaServer.Args(serverTransport);
  if (workerThreads > 0) {
    // Could support the min & max threads, avoiding to preserve existing functionality.
    serverArgs.minWorkerThreads(workerThreads).maxWorkerThreads(workerThreads);
  }
  ExecutorService executorService = createExecutor(
      workerThreads, maxCallQueueSize, metrics);
  serverArgs.executorService(executorService);
  serverArgs.processor(processor);
  serverArgs.transportFactory(transportFactory);
  serverArgs.protocolFactory(protocolFactory);
  return new THsHaServer(serverArgs);
}

private static TServer getTThreadedSelectorServer(TProtocolFactory protocolFactory,
    TProcessor processor, TTransportFactory transportFactory,
    int workerThreads, int selectorThreads, int maxCallQueueSize,
    InetSocketAddress inetSocketAddress, ThriftMetrics metrics)
    throws TTransportException {
  TNonblockingServerTransport serverTransport = new TNonblockingServerSocket(inetSocketAddress);
  log.info("starting HBase ThreadedSelector Thrift server on " + inetSocketAddress.toString());
  TThreadedSelectorServer.Args serverArgs = new TThreadedSelectorServer.Args(serverTransport);
  if (workerThreads > 0) {
    serverArgs.workerThreads(workerThreads);
  }
  if (selectorThreads > 0) {
    serverArgs.selectorThreads(selectorThreads);
  }

  ExecutorService executorService = createExecutor(
      workerThreads, maxCallQueueSize, metrics);
  serverArgs.executorService(executorService);
  serverArgs.processor(processor);
  serverArgs.transportFactory(transportFactory);
  serverArgs.protocolFactory(protocolFactory);
  return new TThreadedSelectorServer(serverArgs);
}

private TServer getServer(int workerThreads, int selectorThreads, int maxCallQueueSize,
      int readTimeout, int backlog, boolean nonblocking, boolean hsha, boolean selector,
      ThriftMetrics metrics, TProtocolFactory protocolFactory, TProcessor processor,
      TTransportFactory transportFactory, InetSocketAddress inetSocketAddress)
        throws TTransportException {
  TServer server;

  if (nonblocking) {
    server = getTNonBlockingServer(protocolFactory, processor, transportFactory,
            inetSocketAddress);
  } else if (hsha) {
    server = getTHsHaServer(protocolFactory, processor, transportFactory, workerThreads,
            maxCallQueueSize, inetSocketAddress, metrics);
  } else if (selector) {
    server = getTThreadedSelectorServer(protocolFactory, processor, transportFactory,
            workerThreads, selectorThreads, maxCallQueueSize, inetSocketAddress, metrics);
  } else {
    server = getTThreadPoolServer(protocolFactory, processor, transportFactory, workerThreads,
            inetSocketAddress, backlog, readTimeout, metrics);
  }
  return server;
}

public static TServer createThreadedSelectorServer(TProcessorFactory processorFactory,
        int port, int clientTimeoutMillisecs, int maxFrameSize, long maxReadBufferSize)
        throws TTransportException {
    int numThreads = Math.max(2, Runtime.getRuntime().availableProcessors());
    int selectorThreads = Math.max(2, Runtime.getRuntime().availableProcessors() / 2);
    TNonblockingServerTransport transport = new TNonblockingServerSocket(port,
            clientTimeoutMillisecs);
    TProtocolFactory protocolFactory = new TBinaryProtocol.Factory();
    TTransportFactory transportFactory = new TFramedTransport.Factory(maxFrameSize);
    TThreadedSelectorServer.Args args = new TThreadedSelectorServer.Args(transport)
            .processorFactory(processorFactory).protocolFactory(protocolFactory)
            .transportFactory(transportFactory).workerThreads(numThreads)
            .acceptPolicy(AcceptPolicy.FAIR_ACCEPT).acceptQueueSizePerThread(10000)
            .selectorThreads(selectorThreads);
    args.maxReadBufferBytes = maxReadBufferSize;
    TThreadedSelectorServer server = new TThreadedSelectorServer(args);
    return server;
}

public TServer getServer(int port, TProcessor processor) throws IOException, TTransportException {
    TTransportFactory serverTransportFactory = getServerTransportFactory();

    //define THsHaServer args 
    //original: THsHaServer + TNonblockingServerSocket
    //option: TThreadPoolServer + TServerSocket
    TServerSocket serverTransport = new TServerSocket(port);
    TThreadPoolServer.Args server_args = new TThreadPoolServer.Args(serverTransport).
            processor(new TUGIWrapProcessor(processor)).
            minWorkerThreads(64).
            maxWorkerThreads(64).
            protocolFactory(new TBinaryProtocol.Factory());            
    if (serverTransportFactory != null) 
        server_args.transportFactory(serverTransportFactory);

    //construct THsHaServer
    return new TThreadPoolServer(server_args);
}

private static TTransportFactory validateAndSetTransportFactory(String transportFactory)
{
    try
    {
        Class factory = Class.forName(transportFactory);

        if(!TTransportFactory.class.isAssignableFrom(factory))
            throw new IllegalArgumentException(String.format("transport factory '%s' " +
                                                             "not derived from TTransportFactory", transportFactory));

        return (TTransportFactory) factory.newInstance();
    }
    catch (Exception e)
    {
        throw new IllegalArgumentException(String.format("Cannot create a transport factory '%s'.", transportFactory), e);
    }
}

private TTransportFactory validateAndSetTransportFactory(String transportFactory)
{
    try
    {
        Class factory = Class.forName(transportFactory);

        if(!TTransportFactory.class.isAssignableFrom(factory))
            throw new IllegalArgumentException(String.format("transport factory '%s' " +
                    "not derived from TTransportFactory", transportFactory));

        return (TTransportFactory) factory.newInstance();
    }
    catch (Exception e)
    {
        throw new IllegalArgumentException(String.format("Cannot create a transport factory '%s'.", transportFactory), e);
    }
}

public ThriftServerThread(InetAddress listenAddr,
                          int listenPort,
                          int listenBacklog,
                          TProcessor processor,
                          TTransportFactory transportFactory)
{
    // now we start listening for clients
    logger.info(String.format("Binding thrift service to %s:%s", listenAddr, listenPort));

    TServerFactory.Args args = new TServerFactory.Args();
    args.tProtocolFactory = new TBinaryProtocol.Factory(true, true);
    args.addr = new InetSocketAddress(listenAddr, listenPort);
    args.listenBacklog = listenBacklog;
    args.processor = processor;
    args.keepAlive = DatabaseDescriptor.getRpcKeepAlive();
    args.sendBufferSize = DatabaseDescriptor.getRpcSendBufferSize();
    args.recvBufferSize = DatabaseDescriptor.getRpcRecvBufferSize();
    args.inTransportFactory = transportFactory;
    args.outTransportFactory = transportFactory;
    serverEngine = new TServerCustomFactory(DatabaseDescriptor.getRpcServerType()).buildTServer(args);
}

public ThriftServerThread(InetAddress listenAddr,
                          int listenPort,
                          int listenBacklog,
                          CassandraServer server,
                          TProcessor processor,
                          TTransportFactory transportFactory)
{
    // now we start listening for clients
    logger.info(String.format("Binding thrift service to %s:%s", listenAddr, listenPort));

    TServerFactory.Args args = new TServerFactory.Args();
    args.tProtocolFactory = new TBinaryProtocol.Factory(true, true);
    args.addr = new InetSocketAddress(listenAddr, listenPort);
    args.listenBacklog = listenBacklog;
    args.cassandraServer = server;
    args.processor = processor;
    args.keepAlive = DatabaseDescriptor.getRpcKeepAlive();
    args.sendBufferSize = DatabaseDescriptor.getRpcSendBufferSize();
    args.recvBufferSize = DatabaseDescriptor.getRpcRecvBufferSize();
    args.inTransportFactory = transportFactory;
    args.outTransportFactory = transportFactory;
    serverEngine = new TServerCustomFactory(DatabaseDescriptor.getRpcServerType()).buildTServer(args);
}

public SettingsTransport(TOptions options)
{
    if (options instanceof SSLOptions)
    {
        throw new UnsupportedOperationException();
    }
    else
    {
        this.fqFactoryClass = options.factory.value();
        try
        {
            Class<?> clazz = Class.forName(fqFactoryClass);
            if (!TTransportFactory.class.isAssignableFrom(clazz))
                throw new ClassCastException();
            // check we can instantiate it
            clazz.newInstance();
        }
        catch (Exception e)
        {
            throw new IllegalArgumentException("Invalid transport factory class: " + options.factory.value(), e);
        }

    }
}

private static TTransportFactory validateAndSetTransportFactory(String transportFactory)
{
    try
    {
        Class factory = Class.forName(transportFactory);

        if(!TTransportFactory.class.isAssignableFrom(factory))
            throw new IllegalArgumentException(String.format("transport factory '%s' " +
                                                             "not derived from TTransportFactory", transportFactory));

        return (TTransportFactory) factory.newInstance();
    }
    catch (Exception e)
    {
        throw new IllegalArgumentException(String.format("Cannot create a transport factory '%s'.", transportFactory), e);
    }
}

private TTransportFactory validateAndSetTransportFactory(String transportFactory)
{
    try
    {
        Class factory = Class.forName(transportFactory);

        if(!TTransportFactory.class.isAssignableFrom(factory))
            throw new IllegalArgumentException(String.format("transport factory '%s' " +
                    "not derived from TTransportFactory", transportFactory));

        return (TTransportFactory) factory.newInstance();
    }
    catch (Exception e)
    {
        throw new IllegalArgumentException(String.format("Cannot create a transport factory '%s'.", transportFactory), e);
    }
}

public TServer getServer(int port, TProcessor processor) throws IOException, TTransportException {
    TTransportFactory serverTransportFactory = getServerTransportFactory();

    //define THsHaServer args 
    //original: THsHaServer + TNonblockingServerSocket
    //option: TThreadPoolServer + TServerSocket
    TServerSocket serverTransport = new TServerSocket(port);
    TThreadPoolServer.Args server_args = new TThreadPoolServer.Args(serverTransport).
            processor(new TUGIWrapProcessor(processor)).
            minWorkerThreads(64).
            maxWorkerThreads(64).
            protocolFactory(new TBinaryProtocol.Factory());            
    if (serverTransportFactory != null) 
        server_args.transportFactory(serverTransportFactory);

    //construct THsHaServer
    return new TThreadPoolServer(server_args);
}

public ThriftServerThread(InetAddress listenAddr,
                          int listenPort,
                          int listenBacklog,
                          TProcessor processor,
                          TTransportFactory transportFactory)
{
    // now we start listening for clients
    logger.info(String.format("Binding thrift service to %s:%s", listenAddr, listenPort));

    TServerFactory.Args args = new TServerFactory.Args();
    args.tProtocolFactory = new TBinaryProtocol.Factory(true, true);
    args.addr = new InetSocketAddress(listenAddr, listenPort);
    args.listenBacklog = listenBacklog;
    args.processor = processor;
    args.keepAlive = DatabaseDescriptor.getRpcKeepAlive();
    args.sendBufferSize = DatabaseDescriptor.getRpcSendBufferSize();
    args.recvBufferSize = DatabaseDescriptor.getRpcRecvBufferSize();
    args.inTransportFactory = transportFactory;
    args.outTransportFactory = transportFactory;
    serverEngine = new TServerCustomFactory(DatabaseDescriptor.getRpcServerType()).buildTServer(args);
}

private static TTransportFactory validateAndSetTransportFactory(String transportFactory)
{
    try
    {
        Class factory = Class.forName(transportFactory);

        if(!TTransportFactory.class.isAssignableFrom(factory))
            throw new IllegalArgumentException(String.format("transport factory '%s' " +
                                                             "not derived from TTransportFactory", transportFactory));

        return (TTransportFactory) factory.newInstance();
    }
    catch (Exception e)
    {
        throw new IllegalArgumentException(String.format("Cannot create a transport factory '%s'.", transportFactory), e);
    }
}

private TTransportFactory validateAndSetTransportFactory(String transportFactory)
{
    try
    {
        Class factory = Class.forName(transportFactory);

        if(!TTransportFactory.class.isAssignableFrom(factory))
            throw new IllegalArgumentException(String.format("transport factory '%s' " +
                    "not derived from TTransportFactory", transportFactory));

        return (TTransportFactory) factory.newInstance();
    }
    catch (Exception e)
    {
        throw new IllegalArgumentException(String.format("Cannot create a transport factory '%s'.", transportFactory), e);
    }
}

/**
 * Start the SessionDriver callback server
 * @param conf the corona configuration for this session
 * @return the server socket of the callback server
 * @throws IOException
 */
private ServerSocket initializeServer(CoronaConf conf) throws IOException {
  // Choose any free port.
  ServerSocket sessionServerSocket =
    new ServerSocket(0, 0, getLocalAddress());
  TServerSocket tServerSocket = new TServerSocket(sessionServerSocket,
                                                  conf.getCMSoTimeout());

  TFactoryBasedThreadPoolServer.Args args =
    new TFactoryBasedThreadPoolServer.Args(tServerSocket);
  args.processor(new SessionDriverServiceProcessor(incoming));
  args.transportFactory(new TTransportFactory());
  args.protocolFactory(new TBinaryProtocol.Factory(true, true));
  args.stopTimeoutVal = 0;
  server = new TFactoryBasedThreadPoolServer(
    args, new TFactoryBasedThreadPoolServer.DaemonThreadFactory());
  return sessionServerSocket;
}

@Override
public TServer getServer(TProcessor processor) throws IOException, TTransportException {
    int port = type.getPort(storm_conf);
    TTransportFactory serverTransportFactory = getServerTransportFactory();
    TServerSocket serverTransport = new TServerSocket(port);
    int numWorkerThreads = type.getNumThreads(storm_conf);
    Integer queueSize = type.getQueueSize(storm_conf);

    TThreadPoolServer.Args server_args =
            new TThreadPoolServer.Args(serverTransport).processor(new TUGIWrapProcessor(processor)).minWorkerThreads(numWorkerThreads)
                    .maxWorkerThreads(numWorkerThreads).protocolFactory(new TBinaryProtocol.Factory(false, true));

    if (serverTransportFactory != null) {
        server_args.transportFactory(serverTransportFactory);
    }
    BlockingQueue workQueue = new SynchronousQueue();
    if (queueSize != null) {
        workQueue = new ArrayBlockingQueue(queueSize);
    }
    ThreadPoolExecutor executorService = new ExtendedThreadPoolExecutor(numWorkerThreads, numWorkerThreads, 60, TimeUnit.SECONDS, workQueue);
    server_args.executorService(executorService);
    return new TThreadPoolServer(server_args);
}

private static int testProcessor(TProcessor processor, List<ToIntFunction<HostAndPort>> clients)
        throws Exception
{
    try (TServerSocket serverTransport = new TServerSocket(0)) {
        TProtocolFactory protocolFactory = new Factory();
        TTransportFactory transportFactory = new TFramedTransport.Factory();
        TServer server = new TSimpleServer(new Args(serverTransport)
                .protocolFactory(protocolFactory)
                .transportFactory(transportFactory)
                .processor(processor));

        Thread serverThread = new Thread(server::serve);
        try {
            serverThread.start();

            int localPort = serverTransport.getServerSocket().getLocalPort();
            HostAndPort address = HostAndPort.fromParts("localhost", localPort);

            int sum = 0;
            for (ToIntFunction<HostAndPort> client : clients) {
                sum += client.applyAsInt(address);
            }
            return sum;
        }
        finally {
            server.stop();
            serverThread.interrupt();
        }
    }
}

private static int testProcessor(TProcessor processor, List<ToIntFunction<HostAndPort>> clients)
        throws Exception
{
    try (TServerSocket serverTransport = new TServerSocket(0)) {
        TProtocolFactory protocolFactory = new TBinaryProtocol.Factory();
        TTransportFactory transportFactory = new TFramedTransport.Factory();
        TServer server = new TSimpleServer(new Args(serverTransport)
                .protocolFactory(protocolFactory)
                .transportFactory(transportFactory)
                .processor(processor));

        Thread serverThread = new Thread(server::serve);
        try {
            serverThread.start();

            int localPort = serverTransport.getServerSocket().getLocalPort();
            HostAndPort address = HostAndPort.fromParts("localhost", localPort);

            int sum = 0;
            for (ToIntFunction<HostAndPort> client : clients) {
                sum += client.applyAsInt(address);
            }
            return sum;
        }
        finally {
            server.stop();
            serverThread.interrupt();
        }
    }
}

private static TServer getTNonBlockingServer(TProtocolFactory protocolFactory, TProcessor processor,
    TTransportFactory transportFactory, InetSocketAddress inetSocketAddress) throws TTransportException {
  TNonblockingServerTransport serverTransport = new TNonblockingServerSocket(inetSocketAddress);
  log.info("starting HBase Nonblocking Thrift server on " + inetSocketAddress.toString());
  TNonblockingServer.Args serverArgs = new TNonblockingServer.Args(serverTransport);
  serverArgs.processor(processor);
  serverArgs.transportFactory(transportFactory);
  serverArgs.protocolFactory(protocolFactory);
  return new TNonblockingServer(serverArgs);
}

private static TServer getTHsHaServer(TProtocolFactory protocolFactory,
    TProcessor processor, TTransportFactory transportFactory,
    int workerThreads,
    InetSocketAddress inetSocketAddress, ThriftMetrics metrics)
    throws TTransportException {
  TNonblockingServerTransport serverTransport = new TNonblockingServerSocket(inetSocketAddress);
  log.info("starting HBase HsHA Thrift server on " + inetSocketAddress.toString());
  THsHaServer.Args serverArgs = new THsHaServer.Args(serverTransport);
  if (workerThreads > 0) {
    // Could support the min & max threads, avoiding to preserve existing functionality.
    serverArgs.minWorkerThreads(workerThreads).maxWorkerThreads(workerThreads);
  }
  ExecutorService executorService = createExecutor(
      workerThreads, metrics);
  serverArgs.executorService(executorService);
  serverArgs.processor(processor);
  serverArgs.transportFactory(transportFactory);
  serverArgs.protocolFactory(protocolFactory);
  return new THsHaServer(serverArgs);
}

/**
 * Start extension by communicating with osquery core and starting thrift
 * server
 * 
 * @param name
 *            name of extension
 * @param version
 *            version of extension
 * @param sdkVersion
 *            version of the osquery SDK used to build this extension
 * @param minSdkVersion
 *            minimum version of the osquery SDK that you can use
 * @throws IOException
 * @throws ExtensionException
 */
public void startExtension(String name, String version, String sdkVersion, String minSdkVersion)
        throws IOException, ExtensionException {
    ExtensionManager.Client client = new ClientManager(EXTENSION_SOCKET).getClient();
    InternalExtensionInfo info = new InternalExtensionInfo(name, version, sdkVersion, minSdkVersion);
    try {
        ExtensionStatus status = client.registerExtension(info, registry);
        if (status.getCode() == 0) {
            this.uuid = status.uuid;
            Processor<PluginManager> processor = new Processor<PluginManager>(this);
            String serverSocketPath = EXTENSION_SOCKET + "." + String.valueOf(uuid);
            File socketFile = new File(serverSocketPath);
            if (socketFile.exists()) {
                socketFile.delete();
            }
            AFUNIXServerSocket socket = AFUNIXServerSocket.bindOn(new AFUNIXSocketAddress(socketFile));
            socketFile.setExecutable(true, false);
            socketFile.setWritable(true, false);
            socketFile.setReadable(true, false);
            TServerSocket transport = new TServerSocket(socket);
            TTransportFactory transportFactory = new TTransportFactory();
            TProtocolFactory protocolFactory = new TBinaryProtocol.Factory();
            TServer server = new TSimpleServer(new Args(transport).processor(processor)
                    .transportFactory(transportFactory).protocolFactory(protocolFactory));

            // Run it
            System.out.println("Starting the server...");
            server.serve();
        } else {
            throw new ExtensionException(1, status.getMessage(), uuid);
        }
    } catch (TException e) {
        throw new ExtensionException(1, "Could not connect to socket", uuid);
    }
}

public void init() {
    try {
        TMultiplexedProcessor processor = new TMultiplexedProcessor();
        for (ServiceArgs service : serverArgs.getServices()) {
            String className = service.getService();
            if (className.endsWith("$Processor")) {
                className = className.substring(0, className.indexOf("$Processor"));
            }
            processor.registerProcessor(className, service.getProcessor());
        }
        if (serverArgs.getNettyServerArgs() != null) {
            this.server = new TNettyServer(serverArgs.getNettyServerArgs().ip(serverArgs.getHost()).port(serverArgs.getPort()));
        } else {
            TNonblockingServerSocket serverTransport = new TNonblockingServerSocket(new InetSocketAddress(serverArgs.getHost(), serverArgs.getPort()));
            //异步IO，需要使用TFramedTransport，它将分块缓存读取。
            TTransportFactory transportFactory = new TFramedTransport.Factory();
            //使用高密度二进制协议
            TProtocolFactory proFactory = new TBinaryProtocol.Factory();
            // Use this for a multithreaded key
            this.server = new TThreadedSelectorServer(new
                    TThreadedSelectorServer.Args(serverTransport)
                    .transportFactory(transportFactory)
                    .protocolFactory(proFactory)
                    .processor(processor)
            );
        }
        log.info("Starting the Thrift key...");
        this.server.setServerEventHandler(new TrpcRegistryEventHandler(serverArgs));
        this.server.serve();
        if (this.serverArgs.getNettyServerArgs() != null) {
            ((TNettyServer) this.server).waitForClose();
        }
    } catch (Exception e) {
        log.error("publish thrift key error", e);
    }
}

@Test
public void testCustomSocketPool() {
    int serverPort = 38003;
    ThriftClientConfiguration configuration = new ThriftClientConfiguration();
    configuration.setTransportFactory(new TTransportFactory());
    configuration.setSocketPool(new TestSocketPoolImpl());
    Factory factory = new GeneralFactory(configuration);
    try (ThriftClient thriftClient = factory.getThriftClient(LOCAL_HOST, serverPort);
            TTransport transport = thriftClient.getTransport()) {
        assertNull(transport);
    } catch (Exception e) {
        fail();
    }
}

private static TTransportFactory getTTransportFactory(boolean framed) {
  if (framed) {
    log.debug("Using framed transport");
    return new TFramedTransport.Factory();
  } else {
    return new TTransportFactory();
  }
}

private static TServer getTNonBlockingServer(TProtocolFactory protocolFactory, THBaseService.Processor processor,
    TTransportFactory transportFactory, InetSocketAddress inetSocketAddress) throws TTransportException {
  TNonblockingServerTransport serverTransport = new TNonblockingServerSocket(inetSocketAddress);
  log.info("starting HBase Nonblocking Thrift server on " + inetSocketAddress.toString());
  TNonblockingServer.Args serverArgs = new TNonblockingServer.Args(serverTransport);
  serverArgs.processor(processor);
  serverArgs.transportFactory(transportFactory);
  serverArgs.protocolFactory(protocolFactory);
  return new TNonblockingServer(serverArgs);
}