HttpSender implements EventSender { @Override public void send(final File file, final CallbackHandler handler) { log.info("Sending local file to collector: {}", file.getAbsolutePath()); final long startTime = System.nanoTime(); final AsyncCompletionHandler<Response> asyncCompletionHandler = new AsyncCompletionHandler<Response>() { @Override public Response onCompleted(final Response response) { activeRequests.decrementAndGet(); if (response.getStatusCode() == 202) { handler.onSuccess(file); } else { handler.onError(new IOException(String.format("Received response %d: %s", response.getStatusCode(), response.getStatusText())), file); } sendTimer.update(System.nanoTime() - startTime, TimeUnit.NANOSECONDS); return response; } @Override public void onThrowable(final Throwable t) { activeRequests.decrementAndGet(); handler.onError(t, file); } }; final HttpJob job = new HttpJob(client, file, asyncCompletionHandler); workers.offer(job); } HttpSender(final String collectorHost, final int collectorPort, final EventType eventType,
                      final long httpMaxWaitTimeInMillis, final long httpMaxKeepAliveInMillis, final int httpWorkersPoolSize);   HttpSender(final ThreadSafeAsyncHttpClient client, final long httpMaxWaitTimeInMillis, final Timer sendTimer, final int httpWorkersPoolSize); @Override void send(final File file, final CallbackHandler handler); @Override synchronized void close();  }
@Test(groups = "slow") public void testSend() throws Exception { logger.info("sending"); sender.send(eventsFile, failureCallbackHandler); Thread.sleep((long) 100); server.start(); logger.info("Started server"); logger.info("sending"); sender.send(eventsFile, successCallbackHandler); Thread.sleep((long) 500); server.stop(); }
@Test(groups = "slow") public void test404() throws Exception { errorServer.start(); logger.info("sending"); sender.send(eventsFile, failureCallbackHandler); Thread.sleep((long) 500); errorServer.stop(); }