Thrift

// TScoket.java
public void open() throws TTransportException {
  if (isOpen()) {
    throw new TTransportException(TTransportException.ALREADY_OPEN, "Socket already connected.");
  }

  if (host_.length() == 0) {
    throw new TTransportException(TTransportException.NOT_OPEN, "Cannot open null host.");
  }
  if (port_ <= 0) {
    throw new TTransportException(TTransportException.NOT_OPEN, "Cannot open without port.");
  }

  if (socket_ == null) {
    initSocket();
  }

  try {
    socket_.connect(new InetSocketAddress(host_, port_), connectTimeout_);
    inputStream_ = new BufferedInputStream(socket_.getInputStream(), 1024);
    outputStream_ = new BufferedOutputStream(socket_.getOutputStream(), 1024);
  } catch (IOException iox) {
    close();
    throw new TTransportException(TTransportException.NOT_OPEN, iox);
  }
}

public void serve() {
  try {
    serverTransport_.listen();
  } catch (TTransportException ttx) {
    LOGGER.error("Error occurred during listening.", ttx);
    return;
  }

  // Run the preServe event
  if (eventHandler_ != null) {
    eventHandler_.preServe();
  }

  setServing(true);

  while (!stopped_) {
    TTransport client = null;
    TProcessor processor = null;
    TTransport inputTransport = null;
    TTransport outputTransport = null;
    TProtocol inputProtocol = null;
    TProtocol outputProtocol = null;
    ServerContext connectionContext = null;
    try {
      client = serverTransport_.accept();
      if (client != null) {
        processor = processorFactory_.getProcessor(client);
        inputTransport = inputTransportFactory_.getTransport(client);
        outputTransport = outputTransportFactory_.getTransport(client);
        inputProtocol = inputProtocolFactory_.getProtocol(inputTransport);
        outputProtocol = outputProtocolFactory_.getProtocol(outputTransport);
        if (eventHandler_ != null) {
          connectionContext = eventHandler_.createContext(inputProtocol, outputProtocol);
        }
        while (true) {
          if (eventHandler_ != null) {
            eventHandler_.processContext(connectionContext, inputTransport, outputTransport);
          }
          if(!processor.process(inputProtocol, outputProtocol)) {
            break;
          }
        }
      }
    } catch (TTransportException ttx) {
      // Client died, just move on
    } catch (TException tx) {
      if (!stopped_) {
        LOGGER.error("Thrift error occurred during processing of message.", tx);
      }
    } catch (Exception x) {
      if (!stopped_) {
        LOGGER.error("Error occurred during processing of message.", x);
      }
    }

    if (eventHandler_ != null) {
      eventHandler_.deleteContext(connectionContext, inputProtocol, outputProtocol);
    }

    if (inputTransport != null) {
      inputTransport.close();
    }

    if (outputTransport != null) {
      outputTransport.close();
    }

  }
  setServing(false);
}

public void serve() {
  try {
    serverTransport_.listen();
  } catch (TTransportException ttx) {
    LOGGER.error("Error occurred during listening.", ttx);
    return;
  }

  // Run the preServe event
  if (eventHandler_ != null) {
    eventHandler_.preServe();
  }

  stopped_ = false;
  setServing(true);
  int failureCount = 0;
  while (!stopped_) {
    try {
      TTransport client = serverTransport_.accept();
      WorkerProcess wp = new WorkerProcess(client);

      int retryCount = 0;
      long remainTimeInMillis = requestTimeoutUnit.toMillis(requestTimeout);
      while(true) {
        try {
          executorService_.execute(wp);
          break;
        } catch(Throwable t) {
          if (t instanceof RejectedExecutionException) {
            retryCount++;
            try {
              if (remainTimeInMillis > 0) {
                //do a truncated 20 binary exponential backoff sleep
                long sleepTimeInMillis = ((long) (random.nextDouble() *
                                                  (1L << Math.min(retryCount, 20)))) * beBackoffSlotInMillis;
                sleepTimeInMillis = Math.min(sleepTimeInMillis, remainTimeInMillis);
                TimeUnit.MILLISECONDS.sleep(sleepTimeInMillis);
                remainTimeInMillis = remainTimeInMillis - sleepTimeInMillis;
              } else {
                client.close();
                wp = null;
                LOGGER.warn("Task has been rejected by ExecutorService " + retryCount
                            + " times till timedout, reason: " + t);
                break;
              }
            } catch (InterruptedException e) {
              LOGGER.warn("Interrupted while waiting to place client on executor queue.");
              Thread.currentThread().interrupt();
              break;
            }
          } else if (t instanceof Error) {
            LOGGER.error("ExecutorService threw error: " + t, t);
            throw (Error)t;
          } else {
            //for other possible runtime errors from ExecutorService, should also not kill serve
            LOGGER.warn("ExecutorService threw error: " + t, t);
            break;
          }
        }
      }
    } catch (TTransportException ttx) {
      if (!stopped_) {
        ++failureCount;
        LOGGER.warn("Transport error occurred during acceptance of message.", ttx);
      }
    }
  }

  executorService_.shutdown();

  // Loop until awaitTermination finally does return without a interrupted
  // exception. If we don't do this, then we'll shut down prematurely. We want
  // to let the executorService clear it's task queue, closing client sockets
  // appropriately.
  long timeoutMS = stopTimeoutUnit.toMillis(stopTimeoutVal);
  long now = System.currentTimeMillis();
  while (timeoutMS >= 0) {
    try {
      executorService_.awaitTermination(timeoutMS, TimeUnit.MILLISECONDS);
      break;
    } catch (InterruptedException ix) {
      long newnow = System.currentTimeMillis();
      timeoutMS -= (newnow - now);
      now = newnow;
    }
  }
  setServing(false);
}

private void select() {
  try {
    // wait for io events.
    selector.select();

    // process the io events we received
    Iterator<SelectionKey> selectedKeys = selector.selectedKeys().iterator();
    while (!stopped_ && selectedKeys.hasNext()) {
      SelectionKey key = selectedKeys.next();
      selectedKeys.remove();

      // skip if not valid
      if (!key.isValid()) {
        cleanupSelectionKey(key);
        continue;
      }

      // if the key is marked Accept, then it has to be the server
      // transport.
      if (key.isAcceptable()) {
        handleAccept();
      } else if (key.isReadable()) {
        // deal with reads
        handleRead(key);
      } else if (key.isWritable()) {
        // deal with writes
        handleWrite(key);
      } else {
        LOGGER.warn("Unexpected state in select! " + key.interestOps());
      }
    }
  } catch (IOException e) {
    LOGGER.warn("Got an IOException while selecting!", e);
  }
}

protected void handleRead(SelectionKey key) {
  FrameBuffer buffer = (FrameBuffer) key.attachment();
  if (!buffer.read()) {
    cleanupSelectionKey(key);
    return;
  }

  // if the buffer's frame read is complete, invoke the method.
  if (buffer.isFrameFullyRead()) {
    if (!requestInvoke(buffer)) {
      cleanupSelectionKey(key);
    }
  }
}

protected boolean requestInvoke(FrameBuffer frameBuffer) {
  frameBuffer.invoke();
  return true;
}

protected boolean requestInvoke(FrameBuffer frameBuffer) {
  Runnable invocation = getRunnable(frameBuffer);
  if (invoker != null) {
    try {
      invoker.execute(invocation);
      return true;
    } catch (RejectedExecutionException rx) {
      LOGGER.warn("ExecutorService rejected execution!", rx);
      return false;
    }
  } else {
    // Invoke on the caller's thread
    invocation.run();
    return true;
  }
}

// server
public class Server {
    public static void main(String[] args) throws TTransportException {
        TServerSocket serverTransport = new TServerSocket(12345);
        TBinaryProtocol.Factory protocolFactory = new TBinaryProtocol.Factory();
        DemoServer.Processor<DemoServer.Iface> processor = new DemoServer.Processor<DemoServer.Iface>(
                new DemoServer.Iface() {
                    @Override
                    public EchoResponse echo(EchoRequest request) throws TException {
                        return new EchoResponse(request.input, new BaseResp());
                    }
                });
        TThreadPoolServer.Args tArgs = new TThreadPoolServer.Args(serverTransport)
                .processor(processor)
                .protocolFactory(protocolFactory);
        TServer server = new TThreadPoolServer(tArgs);
        server.serve();
    }
}


// client
public class Client {
    public static void main(String[] args) throws TException {
        TTransport transport = new TSocket("127.0.0.1", 12345);
        DemoServer.Client client = new DemoServer.Client(new TBinaryProtocol(transport));
        transport.open();
        System.out.println(client.echo(new EchoRequest("abc", new Base())).getOutput());;
        transport.close();
    }
}

// server
public class Server {
    public static void main(String[] args) throws TTransportException {
        DemoServer.AsyncProcessor<DemoServer.AsyncIface> processor = new DemoServer.AsyncProcessor<>(new DemoServer.AsyncIface() {
            @Override
            public void echo(EchoRequest request, AsyncMethodCallback resultHandler) throws TException {
                resultHandler.onComplete(new EchoResponse(request.input, new BaseResp()));
            }
        });
        TNonblockingServerSocket serverTransport = new TNonblockingServerSocket(12345);
        TThreadedSelectorServer.Args tArgs = new TThreadedSelectorServer.Args(serverTransport)
                .processor(processor)
                .transportFactory(new TFramedTransport.Factory())
                .protocolFactory(new TBinaryProtocol.Factory())
                .selectorThreads(2)
                .workerThreads(10);
        TServer server = new TThreadedSelectorServer(tArgs);
        server.serve();
    }
}

// client
public class Client {
    public static void main(String[] args) throws IOException, TException, InterruptedException {
        DemoServer.AsyncClient client = new DemoServer.AsyncClient(
                new TBinaryProtocol.Factory(),
                new TAsyncClientManager(),
                new TNonblockingSocket("127.0.0.1", 12345)
        );

        client.echo(new EchoRequest("abc", new Base()), new AsyncMethodCallback<DemoServer.AsyncClient.echo_call>() {
            @Override
            public void onComplete(DemoServer.AsyncClient.echo_call result) {
                EchoResponse resp = null;
                try {
                    resp = result.getResult();
                    System.out.println(resp.getOutput());
                } catch (TException e) {
                    e.printStackTrace();
                }
            }

            @Override
            public void onError(Exception e) {
                System.out.println(e);
            }
        });
        Thread.sleep(1000);
    }
}