org.apache.http.nio.client.methods.HttpAsyncMethods#org.apache.http.nio.protocol.HttpAsyncRequestProducer源码实例Demo

@Advice.OnMethodEnter(suppress = Throwable.class)
private static void onBeforeExecute(@Advice.Argument(value = 0, readOnly = false) HttpAsyncRequestProducer requestProducer,
                                    @Advice.Argument(2) HttpContext context,
                                    @Advice.Argument(value = 3, readOnly = false) FutureCallback futureCallback,
                                    @Advice.Local("span") @Nullable Span span,
                                    @Advice.Local("wrapped") boolean wrapped) {
    if (tracer == null || tracer.getActive() == null) {
        return;
    }
    final AbstractSpan<?> parent = tracer.getActive();
    span = parent.createExitSpan();
    if (span != null) {
        span.withType(HttpClientHelper.EXTERNAL_TYPE)
            .withSubtype(HttpClientHelper.HTTP_SUBTYPE)
            .activate();

        ApacheHttpAsyncClientHelper<HttpAsyncRequestProducer, FutureCallback<?>, HttpContext, HttpRequest> asyncHelper =
            asyncHelperManager.getForClassLoaderOfClass(HttpAsyncRequestProducer.class);
        TextHeaderSetter<HttpRequest> headerSetter = headerSetterHelperClassManager.getForClassLoaderOfClass(HttpRequest.class);
        if (asyncHelper != null && headerSetter != null) {
            requestProducer = asyncHelper.wrapRequestProducer(requestProducer, span, headerSetter);
            futureCallback = asyncHelper.wrapFutureCallback(futureCallback, context, span);
            wrapped = true;
        }
    }
}
 

@Test
public void executeAsyncResponseIsNotPooledIfPoolNotConfigured() {

    // given
    HCHttp.Builder testObjectFactoryBuilder =
            HCHttpTest.createDefaultHttpObjectFactoryBuilder();
    testObjectFactoryBuilder.withPooledResponseBuffers(false);

    HttpClient client = spy(testObjectFactoryBuilder.build().createClient());

    CloseableHttpAsyncClient asyncClient = mockAsyncClient(client);

    BatchRequest request = createDefaultTestBatchRequest();

    // when
    client.executeAsync(request, createMockTestResultHandler());

    // then
    verify(client).getAsyncClient();
    verify(asyncClient).execute(
            any(HttpAsyncRequestProducer.class),
            asyncConsumerCaptor.capture(),
            any(HttpContext.class),
            any(FutureCallback.class));
    assertEquals(BasicAsyncResponseConsumer.class, asyncConsumerCaptor.getValue().getClass());
}
 

@Test
public void executeAsyncDelegatesToConfiguredAsyncClient() {

    // given
    HttpClient client = Mockito.spy(createDefaultTestObject());
    CloseableHttpAsyncClient asyncClient = mockAsyncClient(client);

    BatchRequest request = createDefaultTestBatchRequest();

    // when
    client.executeAsync(request, createMockTestResultHandler());

    // then
    verify(client).getAsyncClient();
    verify(asyncClient).execute(
            any(HttpAsyncRequestProducer.class),
            any(HttpAsyncResponseConsumer.class),
            any(HttpContext.class),
            any(FutureCallback.class));

}
 

private HCResultCallback mockHttpResponseCallback(ResponseHandler<Response> responseHandler, BatchResult batchResult) throws IOException {

        HttpClient client = Mockito.spy(createDefaultTestObject());
        CloseableHttpAsyncClient asyncClient = mockAsyncClient(client);

        BatchRequest request = mock(BatchRequest.class);
        when(request.getURI()).thenReturn(UUID.randomUUID().toString());
        when(request.getHttpMethodName()).thenReturn(BatchRequest.HTTP_METHOD_NAME);
        ItemSource itemSource = mock(ItemSource.class);
        when(itemSource.getSource()).thenReturn(mock(ByteBuf.class));
        when(request.serialize()).thenReturn(itemSource);

        when(responseHandler.deserializeResponse(any(InputStream.class)))
                .thenReturn(batchResult);

        client.executeAsync(request, responseHandler);
        verify(asyncClient).execute(
                any(HttpAsyncRequestProducer.class),
                any(PoolingAsyncResponseConsumer.class),
                any(HttpClientContext.class),
                hcResultCallbackCaptor.capture());

        return hcResultCallbackCaptor.getValue();
    }
 

@Override public <T> Future<T> execute(HttpAsyncRequestProducer requestProducer,
  HttpAsyncResponseConsumer<T> responseConsumer, HttpContext httpContext,
  FutureCallback<T> callback) {

  TraceContext invocationContext = currentTraceContext.get();
  if (invocationContext != null) {
    httpContext.setAttribute(TraceContext.class.getName(), invocationContext);
  }

  return delegate.execute(
    new TracingAsyncRequestProducer(requestProducer, httpContext),
    new TracingAsyncResponseConsumer<>(responseConsumer, httpContext),
    httpContext,
    callback != null && invocationContext != null
      ? new TraceContextFutureCallback<>(callback, currentTraceContext, invocationContext)
      : callback
  );
}
 

public HttpAsyncRequestProducerWrapper with(HttpAsyncRequestProducer delegate, Span span, TextHeaderSetter<HttpRequest> headerSetter) {
    // Order is important due to visibility - write to delegate last on this (initiating) thread
    this.span = span;
    this.headerSetter = headerSetter;
    this.delegate = delegate;
    return this;
}
 

public void executeRequest() {
    final HttpPost request = new HttpPost(mBuilder.mUri);
    for (Map.Entry<String, String> entry : mBuilder.mHeaders.entrySet()) {
        request.addHeader(entry.getKey(), entry.getValue());
    }
    final HttpEntity entity = new StringEntity(mBuilder.mContentInJson, mBuilder.mContentType);
    request.setEntity(entity);
    final HttpAsyncRequestProducer requestProducer = HttpAsyncMethods.create(request);
    this.mHttpClient.execute(requestProducer, ((Builder) mBuilder).mHttpAsyncResponseConsumer,
            ((Builder) mBuilder).mFutureCallback);
}
 

private HttpRequest getHttpRequest(final Object target) {
    try {
        if (!(target instanceof RequestProducerGetter)) {
            return null;
        }
        final HttpAsyncRequestProducer requestProducer = ((RequestProducerGetter) target)._$PINPOINT$_getRequestProducer();
        return requestProducer.generateRequest();
    } catch (Exception e) {
        return null;
    }
}
 

private NameIntValuePair<String> getHost(final Object target) {
    if (target instanceof RequestProducerGetter) {
        final HttpAsyncRequestProducer producer = ((RequestProducerGetter) target)._$PINPOINT$_getRequestProducer();
        final HttpHost httpHost = producer.getTarget();
        if (httpHost != null) {
            return new NameIntValuePair<String>(httpHost.getHostName(), httpHost.getPort());
        }
    }
    return new NameIntValuePair<String>(null, -1);
}
 

@Override
public HttpAsyncRequestProducer wrapRequestProducer(HttpAsyncRequestProducer requestProducer, Span span, TextHeaderSetter<HttpRequest> headerSetter) {
    return requestProducerWrapperObjectPool.createInstance().with(requestProducer, span, headerSetter);
}
 

private Thread baseTest() throws Throwable {
    Object[] allArguments = new Object[] {
        producer,
        consumer,
        httpContext,
        callback
    };
    Class[] types = new Class[] {
        HttpAsyncRequestProducer.class,
        HttpAsyncResponseConsumer.class,
        HttpContext.class,
        FutureCallback.class
    };
    httpAsyncClientInterceptor.beforeMethod(enhancedInstance, null, allArguments, types, null);
    Assert.assertEquals(CONTEXT_LOCAL.get(), httpContext);
    Assert.assertTrue(allArguments[1] instanceof HttpAsyncResponseConsumerWrapper);
    Assert.assertTrue(allArguments[3] instanceof FutureCallbackWrapper);

    sessionRequestConstructorInterceptor.onConstruct(enhancedInstance, null);

    Thread thread = new Thread(new Runnable() {
        @Override
        public void run() {
            try {
                //start local
                completeInterceptor.beforeMethod(enhancedInstance, null, null, null, null);
                //start request
                requestExecutorInterceptor.beforeMethod(enhancedInstance, null, null, null, null);

                HttpAsyncResponseConsumerWrapper consumerWrapper = new HttpAsyncResponseConsumerWrapper(consumer);

                consumerWrapper.responseReceived(response);

                new FutureCallbackWrapper(callback).completed(null);

            } catch (Throwable throwable) {
                throwable.printStackTrace();
            }
        }
    });
    thread.start();
    return thread;
}
 

TracingAsyncRequestProducer(HttpAsyncRequestProducer requestProducer, HttpContext context) {
  this.requestProducer = requestProducer;
  this.context = context;
}
 

HttpAsyncRequestProducer _$PINPOINT$_getRequestProducer();