import org.asynchttpclient.AsyncCompletionHandler; //导入依赖的package包/类
@Override
public <T> Single<T> sendRequest(SqsAction<T> action) {
    Request asyncRequest = action.toHttpRequest(credentialsProvider.getCredentials());
    SingleSubject<T> responseSubject = SingleSubject.create();
    httpClient.executeRequest(asyncRequest, new AsyncCompletionHandler<Response>() {
        @Override
        public Response onCompleted(Response httpResponse) {
            Single.fromCallable(() -> action.parseHttpResponse(httpResponse))
                    .subscribeWith(responseSubject);
            return httpResponse;
        }

        @Override
        public void onThrowable(Throwable throwable) {
            responseSubject.onError(throwable);
        }
    });
    return responseSubject;
}
 

import org.asynchttpclient.AsyncCompletionHandler; //导入依赖的package包/类
public void send(String to, String text) {
    ArrayList<Param> params = new ArrayList<>();
    params.add(new Param("api_key", key));
    params.add(new Param("api_secret", secret));
    params.add(new Param("from", "Blynk"));
    params.add(new Param("to", to));
    params.add(new Param("text", text));

    httpclient.preparePost("https://rest.nexmo.com/sms/json")
            .setQueryParams(params)
            .execute(new AsyncCompletionHandler<Response>() {
                @Override
                public Response onCompleted(org.asynchttpclient.Response response) throws Exception {
                    if (response.getStatusCode() == 200) {
                        SmsResponse smsResponse = smsResponseReader.readValue(response.getResponseBody());
                        if (!smsResponse.messages[0].status.equals("0")) {
                            log.error(smsResponse.messages[0].error);
                        }
                    }
                    return response;
                }
            });


}
 

import org.asynchttpclient.AsyncCompletionHandler; //导入依赖的package包/类
@Test
public void testSendRequestSuccess() throws Exception {
    Single<Object> result = requestSender.sendRequest(actionMock);
    result.test().assertNotComplete();
    ArgumentCaptor<AsyncCompletionHandler> captor = ArgumentCaptor.forClass(AsyncCompletionHandler.class);
    verify(asyncHttpClientMock).executeRequest((Request) any(), captor.capture());
    AsyncCompletionHandler handler = captor.getValue();

    handler.onCompleted(httpResponseMock);
    result.test().assertValue(actionResponse);
}
 

import org.asynchttpclient.AsyncCompletionHandler; //导入依赖的package包/类
@Test
public void testSendRequestOnThrowable() throws Exception {
    Single<Object> result = requestSender.sendRequest(actionMock);
    result.test().assertNotComplete();
    ArgumentCaptor<AsyncCompletionHandler> captor = ArgumentCaptor.forClass(AsyncCompletionHandler.class);
    verify(asyncHttpClientMock).executeRequest((Request) any(), captor.capture());
    AsyncCompletionHandler handler = captor.getValue();

    RuntimeException exception = new RuntimeException("error");
    handler.onThrowable(exception);
    result.test().assertError(exception);
}
 

import org.asynchttpclient.AsyncCompletionHandler; //导入依赖的package包/类
@Test
public void testSendRequestParseFailed() throws Exception {
    RuntimeException exception = new RuntimeException("error");
    when(actionMock.parseHttpResponse(any())).thenThrow(exception);
    Single<Object> result = requestSender.sendRequest(actionMock);
    result.test().assertNotComplete();
    ArgumentCaptor<AsyncCompletionHandler> captor = ArgumentCaptor.forClass(AsyncCompletionHandler.class);
    verify(asyncHttpClientMock).executeRequest((Request) any(), captor.capture());
    AsyncCompletionHandler handler = captor.getValue();
    handler.onCompleted(httpResponseMock);
    result.test().assertError(exception);
}
 

import org.asynchttpclient.AsyncCompletionHandler; //导入依赖的package包/类
public ToyClientMetricsEmitter(TaskMetricEmitter parent, Request request, AsyncCompletionHandler<T> asyncHandler) {
    this.emitter = parent.start(
            "client.method", request.getMethod(),
            "client.url", request.getUrl(),
            "client.host", request.getUri().getHost());
    this.asyncHandler = asyncHandler;
}
 

import org.asynchttpclient.AsyncCompletionHandler; //导入依赖的package包/类
@Test(groups = "standalone", invocationCount = 10, alwaysRun = true)
public void asyncDoGetKeepAliveHandlerTest_channelClosedDoesNotFail() throws Exception {
    try (AsyncHttpClient client = asyncHttpClient()) {
        // Use a l in case the assert fail
        final CountDownLatch l = new CountDownLatch(2);

        final Map<String, Boolean> remoteAddresses = new ConcurrentHashMap<>();

        AsyncCompletionHandler<Response> handler = new AsyncCompletionHandlerAdapter() {

            @Override
            public Response onCompleted(Response response) throws Exception {
                System.out.println("ON COMPLETED INVOKED " + response.getHeader("X-KEEP-ALIVE"));
                try {
                    assertEquals(response.getStatusCode(), 200);
                    remoteAddresses.put(response.getHeader("X-KEEP-ALIVE"), true);
                } finally {
                    l.countDown();
                }
                return response;
            }
        };

        client.prepareGet(getTargetUrl()).execute(handler).get();
        server.stop();
        server.start();
        client.prepareGet(getTargetUrl()).execute(handler);

        if (!l.await(TIMEOUT, TimeUnit.SECONDS)) {
            fail("Timed out");
        }

        assertEquals(remoteAddresses.size(), 2);
    }
}
 

import org.asynchttpclient.AsyncCompletionHandler; //导入依赖的package包/类
/**
 * This test just make sure the hack used to catch disconnected channel
 * under win7 doesn't throw any exception. The onComplete method must be
 * only called once.
 * 
 * @throws Exception if something wrong happens.
 */
@Test(groups = "standalone")
public void win7DisconnectTest() throws Exception {
    final AtomicInteger count = new AtomicInteger(0);

    try (AsyncHttpClient client = asyncHttpClient()) {
        AsyncCompletionHandler<Response> handler = new AsyncCompletionHandlerAdapter() {

            @Override
            public Response onCompleted(Response response) throws Exception {

                count.incrementAndGet();
                StackTraceElement e = new StackTraceElement("sun.nio.ch.SocketDispatcher", "read0", null, -1);
                IOException t = new IOException();
                t.setStackTrace(new StackTraceElement[] { e });
                throw t;
            }
        };

        try {
            client.prepareGet(getTargetUrl()).execute(handler).get();
            fail("Must have received an exception");
        } catch (ExecutionException ex) {
            assertNotNull(ex);
            assertNotNull(ex.getCause());
            assertEquals(ex.getCause().getClass(), IOException.class);
            assertEquals(count.get(), 1);
        }
    }
}
 

import org.asynchttpclient.AsyncCompletionHandler; //导入依赖的package包/类
@Test(groups = "standalone")
public void zeroCopyPostTest() throws IOException, ExecutionException, TimeoutException, InterruptedException, URISyntaxException {
    try (AsyncHttpClient client = asyncHttpClient()) {
        final AtomicBoolean headerSent = new AtomicBoolean(false);
        final AtomicBoolean operationCompleted = new AtomicBoolean(false);

        Response resp = client.preparePost("http://localhost:" + port1 + "/").setBody(SIMPLE_TEXT_FILE).execute(new AsyncCompletionHandler<Response>() {

            public State onHeadersWritten() {
                headerSent.set(true);
                return State.CONTINUE;
            }

            public State onContentWritten() {
                operationCompleted.set(true);
                return State.CONTINUE;
            }

            @Override
            public Response onCompleted(Response response) throws Exception {
                return response;
            }
        }).get();
        assertNotNull(resp);
        assertEquals(resp.getStatusCode(), HttpServletResponse.SC_OK);
        assertEquals(resp.getResponseBody(), SIMPLE_TEXT_FILE_STRING);
        assertTrue(operationCompleted.get());
        assertTrue(headerSent.get());
    }
}
 

import org.asynchttpclient.AsyncCompletionHandler; //导入依赖的package包/类
public PrecipiceFuture<HTTPStatus, Response> makeRequest(RequestBuilder request) {
    request.setUrl(url);

    final PrecipicePromise<HTTPStatus, Response> promise = Asynchronous.acquirePermitsAndPromise(guardRail, 1L);
    client.executeRequest(request, new AsyncCompletionHandler<Void>() {
        @Override
        public Void onCompleted(Response response) throws Exception {
            int httpStatus = response.getStatusCode();
            if (httpStatus < 200 || httpStatus > 299) {
                promise.complete(HTTPStatus.NON_200, response);
            } else {
                promise.complete(HTTPStatus.STATUS_200, response);
            }
            return null;
        }

        @Override
        public void onThrowable(Throwable t) {
            if (t instanceof TimeoutException) {
                promise.completeExceptionally(HTTPStatus.TIMEOUT, t);
            } else {
                promise.completeExceptionally(HTTPStatus.ERROR, t);
            }
        }
    });
    return promise.future();
}
 

import org.asynchttpclient.AsyncCompletionHandler; //导入依赖的package包/类
@Override
public void registerService() {
    stream(serviceRegistryProperties.getServers().split(","))
            .filter(server -> !isEmpty(server))
            .forEach(discoveryServer -> {
                try {
                    LOG.debug("Updating registration of service at '{}'", discoveryServer);
                    httpClient
                            .preparePut(discoveryServer + "/environments/" + edisonApplicationProperties.getEnvironment() + "/" + applicationInfo.name)
                            .setHeader("Content-Type", "application/vnd.otto.edison.links+json")
                            .setHeader("Accept", "application/vnd.otto.edison.links+json")
                            .setBody(
                                    "{\n" +
                                            "   \"groups\":[\"" + edisonApplicationProperties.getGroup() + "\"],\n" +
                                            "   \"expire\":" + serviceRegistryProperties.getExpireAfter() + ",\n" +
                                            "   \"links\":[{\n" +
                                            "      \"rel\":\"http://github.com/otto-de/edison/link-relations/microservice\",\n" +
                                            "      \"href\" : \"" + serviceRegistryProperties.getService() + "\",\n" +
                                            "      \"title\":\"" + applicationInfo.title + "\"\n" +
                                            "   }]  \n" +
                                            "}"
                            )
                            .execute(new AsyncCompletionHandler<Integer>() {
                                @Override
                                public Integer onCompleted(final Response response) throws Exception {
                                    if (response.getStatusCode() < 300) {
                                        LOG.info("Successfully updated registration at " + discoveryServer);
                                    } else {
                                        LOG.warn("Failed to update registration at '{}': Status='{}' '{}'", discoveryServer, response.getStatusCode(), response.getStatusText());
                                    }
                                    return response.getStatusCode();
                                }

                                @Override
                                public void onThrowable(final Throwable t) {
                                    LOG.error("Failed to register at '{}'", discoveryServer, t);
                                }
                            });
                } catch (final Exception e) {
                    LOG.error("Error updating registration", e);
                }
            });
}
 

import org.asynchttpclient.AsyncCompletionHandler; //导入依赖的package包/类
public void send(GCMMessage messageBase, final Map<String, String> tokens, final String uid) {
    if (gcmURI == null) {
        log.error("Error sending push. Google cloud messaging properties not provided.");
        return;
    }

    String message;
    try {
        messageBase.setTitle(title);
        message = messageBase.toJson();
    } catch (JsonProcessingException e) {
        log.error("Error sending push. Wrong message format.");
        return;
    }

    httpclient.preparePost(gcmURI).setHeader("Authorization", apiKey)
            .setHeader(HttpHeaderNames.CONTENT_TYPE, "application/json; charset=utf-8")
            .setBody(message)
            .execute(new AsyncCompletionHandler<Response>() {
                @Override
                public Response onCompleted(Response response) throws Exception {
                    if (response.getStatusCode() == HttpResponseStatus.OK.code()) {
                        GCMResponseMessage gcmResponseMessage =
                                gcmResponseReader.readValue(response.getResponseBody());
                        if (gcmResponseMessage.failure == 1) {
                            String errorMessage =
                                    gcmResponseMessage.results != null && gcmResponseMessage.results.length > 0
                                            ? gcmResponseMessage.results[0].error
                                            : messageBase.getToken();
                            processError(errorMessage, tokens, uid);
                        }
                        return response;
                    }

                    processError(response.getResponseBody(), tokens, uid);
                    return response;
                }

                @Override
                public void onThrowable(Throwable t) {
                    processError(t.getMessage(), tokens, uid);
                }
            });

}
 

import org.asynchttpclient.AsyncCompletionHandler; //导入依赖的package包/类
private void facebookLogin(ChannelHandlerContext ctx, int messageId, String email,
                           String token, Version version) {
    asyncHttpClient.prepareGet(URL + token)
            .execute(new AsyncCompletionHandler<Response>() {
                @Override
                public Response onCompleted(Response response) throws Exception {
                    if (response.getStatusCode() != 200) {
                        String errMessage = response.getResponseBody();
                        if (errMessage != null && errMessage.contains("expired")) {
                            log.warn("Facebook token expired for user {}.", email);
                        } else {
                            log.warn("Error getting facebook token for user {}. Reason : {}", email, errMessage);
                        }
                        ctx.writeAndFlush(notAllowed(messageId), ctx.voidPromise());
                        return response;
                    }

                    try {
                        String responseBody = response.getResponseBody();
                        FacebookTokenResponse facebookTokenResponse =
                                JsonParser.parseFacebookTokenResponse(responseBody);
                        if (email.equalsIgnoreCase(facebookTokenResponse.email)) {
                            User user = holder.userDao.getByName(email, AppNameUtil.BLYNK);
                            if (user == null) {
                                user = holder.userDao.addFacebookUser(email, AppNameUtil.BLYNK);
                            }

                            login(ctx, messageId, user, version);
                        }
                    } catch (Exception e) {
                        log.error("Error during facebook response parsing for user {}. Reason : {}",
                                email, response.getResponseBody());
                        ctx.writeAndFlush(notAllowed(messageId), ctx.voidPromise());
                    }

                    return response;
                }

                @Override
                public void onThrowable(Throwable t) {
                    log.error("Error performing facebook request. Token {} for user {}. Reason : {}",
                            token, email, t.getMessage());
                    ctx.writeAndFlush(notAllowed(messageId), ctx.voidPromise());
                }
            });
}