Java 类org.apache.thrift.TProcessorFactory 实例源码

public static void main(String[] args){
    ExecutorService es = Executors.newFixedThreadPool(2);
    for(int i=0; i<ports.length; i++){
        final int index = i;
        es.execute(new Runnable() {
            @Override
            public void run() {
                try{
                    TNonblockingServerSocket socket = new TNonblockingServerSocket(ports[index]);
                    TestThriftJ.Processor processor = new TestThriftJ.Processor(new QueryImp());
                    TNonblockingServer.Args arg = new TNonblockingServer.Args(socket);
                    arg.protocolFactory(new TBinaryProtocol.Factory());
                    arg.transportFactory(new TFramedTransport.Factory());
                    arg.processorFactory(new TProcessorFactory(processor));
                    TServer server = new TNonblockingServer (arg);

                    logger.info("127.0.0.1:" + ports[index] + " start");
                    server.serve();
                }catch(Exception e){
                    logger.error("127.0.0.1:" + ports[index] + " error");
                }
            }
        });
    }
}

public static void main(String[] args){
    ExecutorService es = Executors.newFixedThreadPool(2);
    for(int i=0; i<ports.length; i++){
        final int index = i;
        es.execute(new Runnable() {
            @Override
            public void run() {
                try{
                    TNonblockingServerSocket socket = new TNonblockingServerSocket(ports[index]);
                    TestThriftJ.Processor processor = new TestThriftJ.Processor(new QueryImp());
                    TNonblockingServer.Args arg = new TNonblockingServer.Args(socket);
                    arg.protocolFactory(new TBinaryProtocol.Factory());
                    arg.transportFactory(new TFramedTransport.Factory());
                    arg.processorFactory(new TProcessorFactory(processor));
                    TServer server = new TNonblockingServer(arg);

                    logger.info("127.0.0.1:" + ports[index] + " start");
                    server.serve();
                }catch(Exception e){
                    logger.error("127.0.0.1:" + ports[index] + " error");
                }
            }
        });
    }
}

@Test
public void testAysncServiceImpl() {
    int serverPort = 49001;
    TProcessor p = new ServiceProcessor(new TestThriftServiceImpl1());
    thriftServerConfiguration.setProcessorFactory(new TProcessorFactory(p));
    Factory factory = new GeneralFactory(thriftServerConfiguration);
    ThriftServer thriftServer = factory.getThriftServer(serverPort, new TestThriftServiceImpl1());
    thriftServer.run();
    try {
        AsyncService service = factory.getAsyncService(new TNonblockingSocket(LOCAL_HOST, serverPort));
        Thread.sleep(500);
        TestCallback1 callback = new TestCallback1();
        service.get(testString1, callback);
        Thread.sleep(1000);
    } catch (Exception e) {
        fail();
    } finally {
        thriftServer.stop();
    }
}

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 ConcreteServer(AnnotateCommunicationService.Iface impl, int port) throws ServerException {
  try {
    this.serverXport = new TNonblockingServerSocket(port);
    // TODO: eval HaHs server?
    final TNonblockingServer.Args args = new TNonblockingServer.Args(this.serverXport);
    args.protocolFactory(new TCompactProtocol.Factory());
    // TODO: eval FastFramedTransport?
    final TFramedTransport.Factory transFactory = new TFramedTransport.Factory(Integer.MAX_VALUE);
    args.transportFactory(transFactory);
    // legitimately do not know type bound here - guessing Iface
    AnnotateCommunicationService.Processor<Iface> proc = new AnnotateCommunicationService.Processor<>(impl);
    args.processorFactory(new TProcessorFactory(proc));
    args.maxReadBufferBytes = Long.MAX_VALUE;

    this.args = args;
    // final TNonblockingServer server = new TNonblockingServer(args);
    this.server = new TNonblockingServer(this.args);
  } catch (TTransportException e) {
    throw new ServerException(e);
  }
}

public static void startRPCServer3(leafServer leafserver , String ip , int port) throws Exception
{
    TProcessor processor = new leafrpc.Processor<leafrpc.Iface>(new RPCService(leafserver));
    //传输通道，非阻塞模式
    InetSocketAddress address = new InetSocketAddress(InetAddress.getByName(ip),port);
    TNonblockingServerSocket serverTransport = new TNonblockingServerSocket(address,10000);
    TNonblockingServer.Args args = new TNonblockingServer.Args(serverTransport);
    args.protocolFactory(new TBinaryProtocol.Factory());
    args.transportFactory(new TFramedTransport.Factory());
    args.processorFactory(new TProcessorFactory(processor));
    TServer server = new TNonblockingServer(args);
    LOG.info("leaf RPCServer(type:TNonblockingServerSocket) start at ip:port : "+ ip +":" + port );
    server.serve();
}

public ThriftHandler(TProcessorFactory processorFactory, TProtocolFactory inProtocolFactory,
                     TProtocolFactory outProtocolFactory, Executor executor) {
    this.processorFactory = processorFactory;
    this.inProtocolFactory = inProtocolFactory;
    this.outProtocolFactory = outProtocolFactory;
    this.userExecutor = executor;
}

public QThreadPoolServer(TProcessorFactory processorFactory, TProtocolFactory protocolFactory, int requestTimeout,
        ExecutorService executorService, int maxWorkerSize, int minWorkerSize, int port) {
    this.processorFactory = processorFactory;
    this.protocolFactory = protocolFactory;
    this.requestTimeout = requestTimeout;
    this.executorService = executorService;
    this.maxWorkerSize = maxWorkerSize;
    this.minWorkerSize = minWorkerSize;
    this.port = port;
}

@Test
public void testAysncMultiplexedServiceImpl() {
    int serverPort = 49002;
    Map<String, TProcessor> processorMap = new HashMap<>();
    processorMap.put("testAysncService1", new ServiceProcessor(new TestThriftServiceImpl1()));
    processorMap.put("testAysncService2", new ServiceProcessor(new TestThriftServiceImpl2()));
    MultiplexedProcessor p = new MultiplexedProcessor(processorMap);
    thriftServerConfiguration.setProcessorFactory(new TProcessorFactory(p));
    Factory factory = new GeneralFactory(thriftServerConfiguration);
    ThriftServer thriftServer = factory.getThriftServer("testAysncMultiplexedService", serverPort, p);
    thriftServer.run();
    try {
        Thread.sleep(500);
        AsyncService service1 = factory.getAsyncService(new TNonblockingSocket(LOCAL_HOST, serverPort),
                "testAysncService1");
        TestCallback1 callback1 = new TestCallback1();
        service1.get(testString1, callback1);
        AsyncService service2 = factory.getAsyncService(new TNonblockingSocket(LOCAL_HOST, serverPort),
                "testAysncService2");
        TestCallback2 callback2 = new TestCallback2();
        service2.get(testString2, callback2);
        Thread.sleep(1000);
    } catch (Exception e) {
        fail();
    } finally {
        thriftServer.stop();
    }
}

@Test
public void testAysncDefaultThriftServerImpl() {
    int serverPort = 39004;
    TProcessor p = new ThriftSimpleService.Processor<Iface>(new ThriftSimpleServiceImpl());
    ThriftServerConfiguration thriftServerConfiguration = new ThriftServerConfiguration();
    thriftServerConfiguration.setProtocolFactory(new TCompactProtocol.Factory());
    thriftServerConfiguration.setProcessorFactory(new TProcessorFactory(p));
    thriftServerConfiguration.setServerArgsAspect(new ServerArgsAspect() {
        @Override
        public TThreadPoolServer.Args TThreadPoolServerArgsAspect(TThreadPoolServer.Args args) {
            args.stopTimeoutVal = 1;
            return args;
        }
    });
    Factory factory = new GeneralFactory(thriftServerConfiguration);
    ThriftServer thriftServer = factory.getThriftServer(serverName, serverPort, p);
    thriftServer.run();
    try {
        ThriftSimpleService.AsyncClient thriftClient = new ThriftSimpleService.AsyncClient(
                new TCompactProtocol.Factory(), new TAsyncClientManager(),
                new TNonblockingSocket(LOCAL_HOST, serverPort));
        Thread.sleep(500);
        TestCallback callback = new TestCallback();
        thriftClient.get(testString, callback);
        Thread.sleep(1000);
    } catch (Exception e) {
        fail();
    } finally {
        thriftServer.stop();
    }
}

public ThriftHandler(TProcessorFactory processorFactory, TProtocolFactory inProtocolFactory,
        TProtocolFactory outProtocolFactory, Executor executor) {
    this.processorFactory = processorFactory;
    this.inProtocolFactory = inProtocolFactory;
    this.outProtocolFactory = outProtocolFactory;
    this.userExecutor = executor;
}

public ThriftFactories(TTransportFactory inputTransportFactory, TTransportFactory outputTransportFactory,
                       TProtocolFactory  inputProtocolFactory,  TProtocolFactory outputProtocolFactory,
                       TProcessorFactory processorFactory, int maxFrameSizeInBytes)
{
    this.inputTransportFactory = inputTransportFactory;
    this.outputTransportFactory = outputTransportFactory;
    this.inputProtocolFactory = inputProtocolFactory;
    this.outputProtocolFactory = outputProtocolFactory;
    this.processorFactory = processorFactory;
    this.maxFrameSizeInBytes = maxFrameSizeInBytes;
}

public static TServer createThreadedServer(TProcessorFactory processorFactory, int port,
        int clientTimeoutMillisecs, int maxFrameSize) throws TTransportException {
    int maxWorkerThreads = Math.max(2, Runtime.getRuntime().availableProcessors());
    TServerTransport transport = new TServerSocket(port, clientTimeoutMillisecs);
    TProtocolFactory protocolFactory = new TBinaryProtocol.Factory();
    TTransportFactory transportFactory = new TFramedTransport.Factory(maxFrameSize);
    TThreadPoolServer.Args args = new TThreadPoolServer.Args(transport)
            .processorFactory(processorFactory).protocolFactory(protocolFactory)
            .transportFactory(transportFactory).minWorkerThreads(1)
            .maxWorkerThreads(maxWorkerThreads);
    TThreadPoolServer server = new TThreadPoolServer(args);
    return server;
}

public static TServer createNonBlockingServer(TProcessorFactory processorFactory, int port,
        int clientTimeoutMillisecs, int maxFrameSize, long maxReadBufferSize)
        throws TTransportException {
    TNonblockingServerTransport transport = new TNonblockingServerSocket(port,
            clientTimeoutMillisecs);
    TProtocolFactory protocolFactory = new TBinaryProtocol.Factory();
    TTransportFactory transportFactory = new TFramedTransport.Factory(maxFrameSize);
    TNonblockingServer.Args args = new TNonblockingServer.Args(transport)
            .processorFactory(processorFactory).protocolFactory(protocolFactory)
            .transportFactory(transportFactory);
    args.maxReadBufferBytes = maxReadBufferSize;
    TNonblockingServer server = new TNonblockingServer(args);
    return server;
}

public static TServer createHaHsServer(TProcessorFactory processorFactory, int port,
        int clientTimeoutMillisecs, int maxFrameSize, long maxReadBufferSize)
        throws TTransportException {
    int numThreads = Math.max(2, Runtime.getRuntime().availableProcessors());
    TNonblockingServerTransport transport = new TNonblockingServerSocket(port,
            clientTimeoutMillisecs);
    TProtocolFactory protocolFactory = new TBinaryProtocol.Factory();
    TTransportFactory transportFactory = new TFramedTransport.Factory(maxFrameSize);
    THsHaServer.Args args = new THsHaServer.Args(transport).processorFactory(processorFactory)
            .protocolFactory(protocolFactory).transportFactory(transportFactory)
            .workerThreads(numThreads).stopTimeoutVal(60).stopTimeoutUnit(TimeUnit.SECONDS);
    args.maxReadBufferBytes = maxReadBufferSize;
    THsHaServer server = new THsHaServer(args);
    return server;
}

public AbstractServerFactory(int port, TProcessor processor, int clientTimeoutMillisecs,
        int maxFrameSize) {
    this.port = port;
    this.processorFactory = new TProcessorFactory(processor);
    this.clientTimeoutMillisecs = clientTimeoutMillisecs;
    this.maxFrameSize = maxFrameSize;
}

public ThriftServerDef(String name, int serverPort, int maxFrameSize, int queuedResponseLimit, TProcessorFactory factory, TProtocolFactory inProtocolFact, TProtocolFactory outProtocolFact, boolean useHeaderTransport, Executor executor) {
    this.name = name;
    this.serverPort = serverPort;
    this.maxFrameSize = maxFrameSize;
    this.queuedResponseLimit = queuedResponseLimit;
    this.processorFactory = factory;
    this.inProtocolFact = inProtocolFact;
    this.outProtocolFact = outProtocolFact;
    this.headerTransport = useHeaderTransport;
    this.executor = executor;
}

public ConcreteZipArchiveFetchServiceThread(FetchCommunicationService.Iface impl, int port) throws TException {
    serverTransport = new TNonblockingServerSocket(port);
    serverArgs = new TNonblockingServer.Args(serverTransport);
    serverArgs.protocolFactory(new TCompactProtocol.Factory());
    serverArgs.transportFactory(new TFramedTransport.Factory(Integer.MAX_VALUE));
    serverArgs.processorFactory(new TProcessorFactory(new FetchCommunicationService.Processor<>(impl)));
    serverArgs.maxReadBufferBytes = Long.MAX_VALUE;
    server = new TNonblockingServer(serverArgs);
}

/**
 * @param impl
 * @param port
 * @throws TException
 */
public SearchServiceWrapper(SearchService.Iface impl, int port) throws TException {
  this.serverXport = new TNonblockingServerSocket(port);
  final TNonblockingServer.Args args = new TNonblockingServer.Args(this.serverXport);
  args.protocolFactory(new TCompactProtocol.Factory());
  final TFramedTransport.Factory transFactory = new TFramedTransport.Factory(Integer.MAX_VALUE);
  args.transportFactory(transFactory);
  SearchService.Processor<Iface> proc = new SearchService.Processor<>(impl);
  args.processorFactory(new TProcessorFactory(proc));
  args.maxReadBufferBytes = Long.MAX_VALUE;

  this.servArgs = args;
  this.server = new TNonblockingServer(this.servArgs);
}

public FetchServiceWrapper(FetchCommunicationService.Iface impl, int port) throws TException {
  this.serverXport = new TNonblockingServerSocket(port);
  final TNonblockingServer.Args args = new TNonblockingServer.Args(this.serverXport);
  args.protocolFactory(new TCompactProtocol.Factory());
  final TFramedTransport.Factory transFactory = new TFramedTransport.Factory(Integer.MAX_VALUE);
  args.transportFactory(transFactory);
  FetchCommunicationService.Processor<Iface> proc = new FetchCommunicationService.Processor<>(impl);
  args.processorFactory(new TProcessorFactory(proc));
  args.maxReadBufferBytes = Long.MAX_VALUE;

  this.servArgs = args;
  this.server = new TNonblockingServer(this.servArgs);
}

public SummarizationServiceWrapper(SummarizationService.Iface impl, int port) throws TException {
  this.serverXport = new TNonblockingServerSocket(port);
  final TNonblockingServer.Args args = new TNonblockingServer.Args(this.serverXport);
  args.protocolFactory(new TCompactProtocol.Factory());
  final TFramedTransport.Factory transFactory = new TFramedTransport.Factory(Integer.MAX_VALUE);
  args.transportFactory(transFactory);
  SummarizationService.Processor<SummarizationService.Iface> proc = new SummarizationService.Processor<>(impl);
  args.processorFactory(new TProcessorFactory(proc));
  args.maxReadBufferBytes = Long.MAX_VALUE;

  this.servArgs = args;
  this.server = new TNonblockingServer(this.servArgs);
}

public StoreServiceWrapper(StoreCommunicationService.Iface impl, int port) throws TException {
  this.serverXport = new TNonblockingServerSocket(port);
  final TNonblockingServer.Args args = new TNonblockingServer.Args(this.serverXport);
  args.protocolFactory(new TCompactProtocol.Factory());
  final TFramedTransport.Factory transFactory = new TFramedTransport.Factory(Integer.MAX_VALUE);
  args.transportFactory(transFactory);
  StoreCommunicationService.Processor<Iface> proc = new StoreCommunicationService.Processor<>(impl);
  args.processorFactory(new TProcessorFactory(proc));
  args.maxReadBufferBytes = Long.MAX_VALUE;

  this.servArgs = args;
  this.server = new TNonblockingServer(this.servArgs);
}

/**
 * Helper method to create a new framed-transport, non-blocking
 * {@link TServer}.
 * 
 * @param processorFactory
 * @param protocolFactory
 * @param port
 * @param clientTimeoutMillisecs
 * @param maxFrameSize
 * @param maxReadBufferSize
 * @return
 * @throws TTransportException
 */
public static TServer createNonBlockingServer(TProcessorFactory processorFactory,
        TProtocolFactory protocolFactory, int port, int clientTimeoutMillisecs,
        int maxFrameSize, long maxReadBufferSize) throws TTransportException {
    TNonblockingServerTransport transport = new TNonblockingServerSocket(port,
            clientTimeoutMillisecs);
    TTransportFactory transportFactory = new TFramedTransport.Factory(maxFrameSize);
    TNonblockingServer.Args args = new TNonblockingServer.Args(transport)
            .processorFactory(processorFactory).protocolFactory(protocolFactory)
            .transportFactory(transportFactory);
    args.maxReadBufferBytes = maxReadBufferSize;
    TNonblockingServer server = new TNonblockingServer(args);
    return server;
}

public Builder(TProcessorFactory processorFactory, int port) {
    this.processorFactory = processorFactory;
    this.port = port;
}

public Builder(TProcessorFactory processorFactory, int port) {
    this.processorFactory = processorFactory;
    this.port = port;
}

public ThriftServerConfiguration(TProcessorFactory processorFactory) {
    this.processorFactory = processorFactory;
}

public void setProcessorFactory(TProcessorFactory processorFactory) {
    this.processorFactory = processorFactory;
}

public TProcessorFactory getProcessorFactory() {
    return processorFactory;
}

/**
 * Creates a {@link TThreadPoolServer} server.
 * 
 * <ul>
 * <li>1 dedicated thread for accepting connections</li>
 * <li>Once a connection is accepted, it gets scheduled to be processed by a
 * worker thread. The worker thread is tied to the specific client
 * connection until it's closed. Once the connection is closed, the worker
 * thread goes back to the thread pool.</li>
 * </ul>
 * 
 * @param processorFactory
 * @param protocolFactory
 * @param port
 *            port number on which the Thrift server will listen
 * @param clientTimeoutMillisecs
 * @param maxFrameSize
 *            max size (in bytes) of a transport frame, supply {@code <=0}
 *            value to let the method choose a default {@code maxFrameSize}
 *            value (which is 1Mb)
 * @param maxWorkerThreads
 *            max number of worker threads, supply {@code <=0} value to let
 *            the method choose a default {@code maxWorkerThreads} value
 *            (which is
 *            {@code Math.max(4, Runtime.getRuntime().availableProcessors())}
 *            )
 * @return
 * @throws TTransportException
 */
public static TThreadPoolServer createThreadPoolServer(TProcessorFactory processorFactory,
        TProtocolFactory protocolFactory, int port, int clientTimeoutMillisecs,
        int maxFrameSize, int maxWorkerThreads) throws TTransportException {
    if (clientTimeoutMillisecs <= 0) {
        clientTimeoutMillisecs = DEFAULT_CLIENT_TIMEOUT_MS;
    }
    if (maxFrameSize <= 0) {
        maxFrameSize = DEFAULT_MAX_FRAMESIZE;
    }
    if (maxWorkerThreads <= 0) {
        maxWorkerThreads = DEFAULT_NUM_WORKER_THREADS;
    }

    TServerTransport transport = new TServerSocket(port, clientTimeoutMillisecs);
    TTransportFactory transportFactory = new TFramedTransport.Factory(maxFrameSize);
    TThreadPoolServer.Args args = new TThreadPoolServer.Args(transport)
            .processorFactory(processorFactory).protocolFactory(protocolFactory)
            .transportFactory(transportFactory).minWorkerThreads(1)
            .maxWorkerThreads(maxWorkerThreads);
    TThreadPoolServer server = new TThreadPoolServer(args);
    return server;
}

/**
 * Creates a {@link TNonblockingServer} server.
 * 
 * <p>
 * Non-blocking Thrift server that uses {@code java.nio.channels.Selector}
 * to handle multiple clients. However, messages are processed by the same
 * thread that calls {@code select()}.
 * </p>
 * 
 * @param processorFactory
 * @param protocolFactory
 * @param port
 *            port number on which the Thrift server will listen
 * @param clientTimeoutMillisecs
 * @param maxFrameSize
 *            max size (in bytes) of a transport frame, supply {@code <=0}
 *            value to let the method choose a default {@code maxFrameSize}
 *            value (which is 1Mb)
 * @param maxReadBufferSize
 *            max size (in bytes) of read buffer, supply {@code <=0} value
 *            to let the method choose a default {@code maxReadBufferSize}
 *            value (which is 16Mb)
 * @return
 * @throws TTransportException
 */
public static TNonblockingServer createNonBlockingServer(TProcessorFactory processorFactory,
        TProtocolFactory protocolFactory, int port, int clientTimeoutMillisecs,
        int maxFrameSize, long maxReadBufferSize) throws TTransportException {
    if (clientTimeoutMillisecs <= 0) {
        clientTimeoutMillisecs = DEFAULT_CLIENT_TIMEOUT_MS;
    }
    if (maxFrameSize <= 0) {
        maxFrameSize = DEFAULT_MAX_FRAMESIZE;
    }
    if (maxReadBufferSize <= 0) {
        maxReadBufferSize = DEFAULT_TOTAL_MAX_READ_BUFFERSIZE;
    }

    TNonblockingServerTransport transport = new TNonblockingServerSocket(port,
            clientTimeoutMillisecs);
    TTransportFactory transportFactory = new TFramedTransport.Factory(maxFrameSize);
    TNonblockingServer.Args args = new TNonblockingServer.Args(transport)
            .processorFactory(processorFactory).protocolFactory(protocolFactory)
            .transportFactory(transportFactory);
    args.maxReadBufferBytes = maxReadBufferSize;
    TNonblockingServer server = new TNonblockingServer(args);
    return server;
}

public T processorFactory(TProcessorFactory factory) {
  this.processorFactory = factory;
  return (T) this;
}

/**
 * Creates a {@link TThreadPoolServer} server.
 * 
 * <ul>
 * <li>1 dedicated thread for accepting connections</li>
 * <li>Once a connection is accepted, it gets scheduled to be processed by a
 * worker thread. The worker thread is tied to the specific client
 * connection until it's closed. Once the connection is closed, the worker
 * thread goes back to the thread pool.</li>
 * </ul>
 * 
 * @param processorFactory
 * @param protocolFactory
 * @param port
 *            port number on which the Thrift server will listen
 * @param clientTimeoutMillisecs
 * @param maxFrameSize
 *            max size (in bytes) of a transport frame, supply {@code <=0}
 *            value to let the method choose a default {@code maxFrameSize}
 *            value (which is 1Mb)
 * @param maxWorkerThreads
 *            max number of worker threads, supply {@code <=0} value to let
 *            the method choose a default {@code maxWorkerThreads} value
 *            (which is
 *            {@code Math.max(2, Runtime.getRuntime().availableProcessors())}
 *            )
 * @return
 * @throws TTransportException
 */
public static TThreadPoolServer createThreadPoolServer(TProcessorFactory processorFactory,
        TProtocolFactory protocolFactory, int port, int clientTimeoutMillisecs,
        int maxFrameSize, int maxWorkerThreads) throws TTransportException {
    if (clientTimeoutMillisecs <= 0) {
        clientTimeoutMillisecs = DEFAULT_CLIENT_TIMEOUT_MS;
    }
    if (maxFrameSize <= 0) {
        maxFrameSize = DEFAULT_MAX_FRAMESIZE;
    }
    if (maxWorkerThreads <= 0) {
        maxWorkerThreads = DEFAULT_NUM_WORKER_THREADS;
    }

    TServerTransport transport = new TServerSocket(port, clientTimeoutMillisecs);
    TTransportFactory transportFactory = new TFramedTransport.Factory(maxFrameSize);
    TThreadPoolServer.Args args = new TThreadPoolServer.Args(transport)
            .processorFactory(processorFactory).protocolFactory(protocolFactory)
            .transportFactory(transportFactory).minWorkerThreads(1)
            .maxWorkerThreads(maxWorkerThreads);
    TThreadPoolServer server = new TThreadPoolServer(args);
    return server;
}

/**
 * Creates a {@link TNonblockingServer} server.
 * 
 * <p>
 * Non-blocking Thrift server that uses {@code java.nio.channels.Selector}
 * to handle multiple clients. However, messages are processed by the same
 * thread that calls {@code select()}.
 * </p>
 * 
 * @param processorFactory
 * @param protocolFactory
 * @param port
 *            port number on which the Thrift server will listen
 * @param clientTimeoutMillisecs
 * @param maxFrameSize
 *            max size (in bytes) of a transport frame, supply {@code <=0}
 *            value to let the method choose a default {@code maxFrameSize}
 *            value (which is 1Mb)
 * @param maxReadBufferSize
 *            max size (in bytes) of read buffer, supply {@code <=0} value
 *            to let the method choose a default {@code maxReadBufferSize}
 *            value (which is 16Mb)
 * @return
 * @throws TTransportException
 */
public static TNonblockingServer createNonBlockingServer(TProcessorFactory processorFactory,
        TProtocolFactory protocolFactory, int port, int clientTimeoutMillisecs,
        int maxFrameSize, long maxReadBufferSize) throws TTransportException {
    if (clientTimeoutMillisecs <= 0) {
        clientTimeoutMillisecs = DEFAULT_CLIENT_TIMEOUT_MS;
    }
    if (maxFrameSize <= 0) {
        maxFrameSize = DEFAULT_MAX_FRAMESIZE;
    }
    if (maxReadBufferSize <= 0) {
        maxReadBufferSize = DEFAULT_TOTAL_MAX_READ_BUFFERSIZE;
    }

    TNonblockingServerTransport transport = new TNonblockingServerSocket(port,
            clientTimeoutMillisecs);
    TTransportFactory transportFactory = new TFramedTransport.Factory(maxFrameSize);
    TNonblockingServer.Args args = new TNonblockingServer.Args(transport)
            .processorFactory(processorFactory).protocolFactory(protocolFactory)
            .transportFactory(transportFactory);
    args.maxReadBufferBytes = maxReadBufferSize;
    TNonblockingServer server = new TNonblockingServer(args);
    return server;
}

protected TProcessorFactory getProcessorFactory() {
    return processorFactory;
}

public AbstractServerFactory setProcessorFactory(TProcessorFactory processorFactory) {
    this.processorFactory = processorFactory;
    return this;
}

public TProcessorFactory getProcessorFactory() {
    return processorFactory;
}

/**
 * Specify the TProcessor.
 */
public ThriftServerDefBuilder withProcessor(TProcessor p) {
    this.processorFactory = new TProcessorFactory(p);
    return this;
}

/**
 * Anohter way to specify the TProcessor.
 */
public ThriftServerDefBuilder withProcessorFactory(TProcessorFactory processorFactory) {
    this.processorFactory = processorFactory;
    return this;
}

public T processorFactory(TProcessorFactory factory) {
  this.processorFactory = factory;
  return (T) this;
}

public T processorFactory(TProcessorFactory factory) {
  this.processorFactory = factory;
  return (T) this;
}