Java 类org.apache.thrift.async.TAsyncClientManager 实例源码

public static Constructor<?> getClientConstructor(Class<?> svcInterface) {
    String client = svcInterface.getName().indexOf("Async") > 0 ? ASYNC_CLIENT_NAME : CLIENT_NAME;
    Class<?>[] args = svcInterface.getName().indexOf("Async") > 0 ? new Class[]{TProtocolFactory.class, TAsyncClientManager.class, TNonblockingTransport.class} : new Class[]{TProtocol.class};

    Class<?> clientClass = getThriftServiceInnerClassOrNull(svcInterface.getEnclosingClass(), client, false);
    if (clientClass == null) {
        throw new ThriftRuntimeException("the client class is null");
    }

    Constructor<?> constructor = ClassUtils.getConstructorIfAvailable(clientClass, args);
    if (constructor == null) {
        throw new ThriftRuntimeException("the clientClass constructor is null");
    }

    return constructor;
}

@Test
@Ignore
public void testAsync() throws TException, IOException, ExecutionException, InterruptedException {
    AskerServer server = new AskerServer(port, false);
    server.start();

    try (SocketChannel socketChannel = SocketChannel.open()) {
        socketChannel.connect(new InetSocketAddress("localhost", port));
        TNonblockingTransport transport = new TNonblockingSocket(socketChannel);

        final Asker.AsyncClient client = new Asker.AsyncClient(
                new TCompactProtocol.Factory(),
                new TAsyncClientManager(), transport);

        Helper helper = new Helper(collector);

        helper.checkEcho(client);
        helper.checkCount(client);
        helper.checkReverse(client);
        helper.checkUpperCast(client);
        helper.checkLowerCast(client);

        helper.checkRandom(client, 5 + random.nextInt(10));
    }
    server.stop();
}

@Override
@SuppressWarnings("unchecked")
public <X extends TAsyncClient> X getClient(final Class<X> clazz) {
    return (X) super.clients.computeIfAbsent(ClassNameUtils.getOuterClassName(clazz), (className) -> {
        TProtocolFactory protocolFactory = (TProtocolFactory) tTransport -> {
            TProtocol protocol = new TBinaryProtocol(tTransport);
            return new TMultiplexedProtocol(protocol, className);
        };
        try {
            return clazz.getConstructor(TProtocolFactory.class, TAsyncClientManager.class, TNonblockingTransport.class)
                    .newInstance(protocolFactory, this.clientManager, this.transport);
        } catch (Throwable e) {
            if (e instanceof UnresolvedAddressException) {
                this.isOpen = false;
            }
            return null;
        }
    });
}

public static Constructor<?> getClientConstructor(Class<?> svcInterface) {
    String client = svcInterface.getName().indexOf("Async") > 0 ? ASYNC_CLIENT_NAME : CLIENT_NAME;
    Class<?>[] args = svcInterface.getName().indexOf("Async") > 0 ? new Class[]{TProtocolFactory.class, TAsyncClientManager.class, TNonblockingTransport.class} : new Class[]{TProtocol.class};

    Class<?> clientClass = getThriftServiceInnerClassOrNull(svcInterface.getEnclosingClass(), client, false);
    if (clientClass == null) {
        throw new ThriftRuntimeException("the client class is null");
    }

    Constructor<?> constructor = ClassUtils.getConstructorIfAvailable(clientClass, args);
    if (constructor == null) {
        throw new ThriftRuntimeException("the clientClass constructor is null");
    }

    return constructor;
}

@Override
public void open() {
    try {
        //异步调用管理器
        this.clientManager = new TAsyncClientManager();
        //设置传输通道，调用非阻塞IO。
        this.transport = new TNonblockingSocket(this.serverInfo.getIp(), this.serverInfo.getPort(), 1000);
    } catch (Exception e) {
        log.error("create AsyncTrpcClient:" + this.serverInfo + " error", e);
        throw new TRpcException("create AsyncTrpcClient:" + this.serverInfo + " error", e);
    }
}

@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 ThriftClientPool(MakerFactory<T> maker) {
  pool = new GenericKeyedObjectPool<InetSocketAddress, T>(new PoolFactory(maker),
      getPoolConfig());
  try {
    clientManager = new TAsyncClientManager();
  } catch (IOException e) {
    LOG.fatal(e);
  }
}

@Override
public void verifyTraces(PluginTestVerifier verifier, String expectedMessage) throws Exception {
    final InetSocketAddress actualServerAddress = this.environment.getServerAddress();
    // ********** Asynchronous Traces
    // SpanEvent - Asynchronous Invocation
    ExpectedTrace asyncInvocationTrace = event("ASYNC", "Asynchronous Invocation");

    // SpanEvent - TAsyncMethodCall.cleanUpAndFireCallback
    Method cleanUpAndFireCallback = TAsyncMethodCall.class.getDeclaredMethod("cleanUpAndFireCallback",
            SelectionKey.class);
    ExpectedTrace cleanUpAndFireCallbackTrace = event("THRIFT_CLIENT_INTERNAL", cleanUpAndFireCallback);

    // SpanEvent - TServiceClient.receiveBase
    Method receiveBase = TServiceClient.class.getDeclaredMethod("receiveBase", TBase.class, String.class);
    ExpectedAnnotation thriftResult = Expectations.annotation("thrift.result", "echo_result(success:"
            + expectedMessage + ")");
    ExpectedTrace receiveBaseTrace = event("THRIFT_CLIENT_INTERNAL", // ServiceType
            receiveBase, // Method
            thriftResult // Annotation("thrift.result")
    );

    // ********** Root trace for Asynchronous traces
    // SpanEvent - TAsyncClientManager.call
    Method call = TAsyncClientManager.class.getDeclaredMethod("call", TAsyncMethodCall.class);
    ExpectedAnnotation thriftUrl = Expectations.annotation("thrift.url", actualServerAddress.getHostName() + ":"
            + actualServerAddress.getPort() + "/com/navercorp/pinpoint/plugin/thrift/dto/EchoService/echo_call");
    ExpectedTrace callTrace = event("THRIFT_CLIENT", // ServiceType
            call, // Method
            null, // rpc
            null, // endPoint
            actualServerAddress.getHostName() + ":" + actualServerAddress.getPort(), // destinationId
            thriftUrl // Annotation("thrift.url")
    );
    verifier.verifyTrace(async(callTrace, asyncInvocationTrace, cleanUpAndFireCallbackTrace, receiveBaseTrace));
}

@Test
public void test_AsyncClient() throws Throwable {
    Random rnd = new Random(System.nanoTime());

    TProtocolFactory[] protfacs = new TProtocolFactory[] { new TCompactProtocol.Factory(),
            new TBinaryProtocol.Factory(), new TJSONProtocol.Factory(),
            new TSimpleJSONProtocol.Factory(TCalculator.Iface.class, false) };

    TProtocolFactory protocolFactory = protfacs[rnd.nextInt(protfacs.length)];

    System.out.println("protocolFactory: " + protocolFactory);

    TAsyncClientManager clientManager = new TAsyncClientManager();
    TNonblockingTransport transport = new TNonblockingSocket(HOST, PORT);
    TCalculator.AsyncClient client = new TCalculator.AsyncClient(protocolFactory, clientManager, transport);
    final int num1 = rnd.nextInt(Integer.MAX_VALUE / 2 - 1);
    final int num2 = rnd.nextInt(Integer.MAX_VALUE / 2 - 1);

    final CountDownLatch latch = new CountDownLatch(1);
    final Throwable[] exceptions = new Throwable[1];
    AsyncMethodCallback<TCalculator.AsyncClient.add_call> resultHandler = new AsyncMethodCallback<TCalculator.AsyncClient.add_call>() {
        @Override
        public void onComplete(TCalculator.AsyncClient.add_call response) {
            System.out.println("onComplete!");
            try {
                int result = response.getResult();
                Assert.assertEquals(num1 + num2, result);
            } catch (Throwable e) {
                exceptions[0] = e;
            } finally {
                latch.countDown();
            }
        }

        @Override
        public void onError(Exception exception) {
            System.err.println("onError!");
            exception.printStackTrace();
            latch.countDown();
        }

    };
    client.add(num1, num2, resultHandler);
    latch.await();
    transport.close();
    if (exceptions[0] != null) {
        throw exceptions[0];
    }
}

public AsyncServiceClientImpl(TProtocolFactory protocolFactory, TAsyncClientManager manager,
        TNonblockingTransport transport) {
    super(protocolFactory, manager, transport);
}

public AsyncServiceClientImpl(TProtocolFactory protocolFactory, TNonblockingTransport transport)
        throws IOException {
    super(protocolFactory, new TAsyncClientManager(), transport);
}

public T create(TNonblockingTransport tr, TAsyncClientManager mgr,
TProtocolFactory factory);

@Override
public InternalService.AsyncClient create(TNonblockingTransport tr,
    TAsyncClientManager mgr, TProtocolFactory factory) {
  return new InternalService.AsyncClient(factory, mgr, tr);
}

@Override
public NodeMonitorService.AsyncClient create(TNonblockingTransport tr,
    TAsyncClientManager mgr, TProtocolFactory factory) {
  return new NodeMonitorService.AsyncClient(factory, mgr, tr);
}

@Override
public SchedulerService.AsyncClient create(TNonblockingTransport tr,
    TAsyncClientManager mgr, TProtocolFactory factory) {
  return new SchedulerService.AsyncClient(factory, mgr, tr);
}

@Override
public FrontendService.AsyncClient create(TNonblockingTransport tr,
    TAsyncClientManager mgr, TProtocolFactory factory) {
  return new FrontendService.AsyncClient(factory, mgr, tr);
}

@Override
public GetTaskService.AsyncClient create(TNonblockingTransport tr,
    TAsyncClientManager mgr, TProtocolFactory factory) {
  return new GetTaskService.AsyncClient(factory, mgr, tr);
}

@Override
public PongService.AsyncClient create(TNonblockingTransport tr,
    TAsyncClientManager mgr, TProtocolFactory factory) {
  return new PongService.AsyncClient(factory, mgr, tr);
}

public static void main(String[] args) 
        throws IOException, InterruptedException, TException {
    //Async client and I/O stack setup
    TNonblockingSocket trans_ep = new TNonblockingSocket("localhost", 9090);
    TAsyncClientManager client_man = new TAsyncClientManager();
    TradeReporting.TradeHistory.AsyncClient client = 
        new TradeReporting.TradeHistory.AsyncClient(new TBinaryProtocol.Factory(),
                                                    client_man, trans_ep);

    //get_last_sale() async callback handler
    WaitableCallback<TradeReport> wc = 
            new WaitableCallback<TradeReport>() {

        @Override
        public void onComplete(TradeReport tr) {
            try {
                System.out.println("[Client] received [" + tr.seq_num + "] " + 
                                   tr.symbol + " : " + tr.size + " @ " + tr.price);
            } finally {
                complete();
            }
        }
    };

    //Make async calls
    wc.reset();
    client.get_last_sale("IBM", wc);
    System.out.println("[Client] get_last_sale() executing asynch...");
    wc.wait(500);
    wc.reset();
    client.get_last_sale("F", wc);
    wc.wait(25000);

    //Make an async call which will time out
    client.setTimeout(1000);
    wc.reset();
    client.get_last_sale("GE", wc);
    wc.wait(5000);

    //Shutdown async client manager and close network socket
    client_man.stop();
    trans_ep.close();
}

public Factory(TAsyncClientManager clientManager, TProtocolFactory protocolFactory) {
  this.clientManager = clientManager;
  this.protocolFactory = protocolFactory;
}

public AsyncClient(TProtocolFactory protocolFactory, TAsyncClientManager clientManager, TNonblockingTransport transport) {
  super(protocolFactory, clientManager, transport);
}

public static ReplicaService.AsyncClient getReplicaServiceAsyncClient(
        String host, int port) throws TTransportException, IOException {
    return new ReplicaService.AsyncClient(new TBinaryProtocol.Factory(),
                                          new TAsyncClientManager(),
                                          new TNonblockingSocket(host, port, Config.getSocketTimeout()));
}

public static AntiEntropyService.AsyncClient getAntiEntropyServiceAsyncClient(
        String host, int port) throws TTransportException, IOException {
    return new AntiEntropyService.AsyncClient(new TBinaryProtocol.Factory(), new TAsyncClientManager(),
                                              new TNonblockingSocket(host, port, Config.getSocketTimeout()));
}

public SampleThriftClient() throws IOException, TException {

        socket = new TNonblockingSocket("localhost", ThriftService.DEFAULT_PORT);
        client = new RainService.AsyncClient(new TBinaryProtocol.Factory(), new TAsyncClientManager(), socket);

        client.startBenchmark(System.currentTimeMillis(), new StartCallback());
    }