@Test public void post_should_post_to_ingestion_endpoint() throws URISyntaxException, InterruptedException { Thread.sleep(1000); stubFor(post(urlEqualTo("/SOURCEID")) .willReturn(aResponse() .withStatus(200) .withHeader("Content-Type", "application/json") .withHeader("X-Ingestion-Data-RateLimit-Reset-Ttl", "1000") .withBody("{\"accepted\":1,\"total_message_bytes\":2}"))); Logger log = mock(Logger.class); DataSiftConfig config = new DataSiftConfig(); Metrics metrics = new Metrics(); config.baseURL = "http: config.sourceID = "SOURCEID"; config.port = 18089; config.username = "USER"; config.apiKey = "APIKEY"; BulkManager bm = new BulkManager(config, null, null, metrics); try { HttpResponse response = bm.post("{}"); String body = EntityUtils.toString(response.getEntity()); assertEquals("{\"accepted\":1,\"total_message_bytes\":2}", body); } catch (Exception e) { e.printStackTrace(); assertNull(e); } }
@SuppressWarnings("checkstyle:designforextension") public HttpResponse post(final String data) throws URISyntaxException, IOException { log.debug("post()"); URI uri = getUri(); String authToken = getAuthorizationToken(config); metrics.sendAttempt.mark(); log.trace("Posting to ingestion endpoint {}", data); log.debug("Posting to ingestion endpoint data length {} bytes", data.length()); Request request = Request.Post(uri) .useExpectContinue() .version(HttpVersion.HTTP_1_1) .addHeader("Auth", authToken) .bodyString(data, ContentType.create("application/json")); return Executor.newInstance(httpClient) .execute(request) .returnResponse(); }
BulkManager implements Runnable { @SuppressWarnings("checkstyle:designforextension") public HttpResponse post(final String data) throws URISyntaxException, IOException { log.debug("post()"); URI uri = getUri(); String authToken = getAuthorizationToken(config); metrics.sendAttempt.mark(); log.trace("Posting to ingestion endpoint {}", data); log.debug("Posting to ingestion endpoint data length {} bytes", data.length()); Request request = Request.Post(uri) .useExpectContinue() .version(HttpVersion.HTTP_1_1) .addHeader("Auth", authToken) .bodyString(data, ContentType.create("application/json")); return Executor.newInstance(httpClient) .execute(request) .returnResponse(); } }
BulkManager implements Runnable { @SuppressWarnings("checkstyle:designforextension") public HttpResponse post(final String data) throws URISyntaxException, IOException { log.debug("post()"); URI uri = getUri(); String authToken = getAuthorizationToken(config); metrics.sendAttempt.mark(); log.trace("Posting to ingestion endpoint {}", data); log.debug("Posting to ingestion endpoint data length {} bytes", data.length()); Request request = Request.Post(uri) .useExpectContinue() .version(HttpVersion.HTTP_1_1) .addHeader("Auth", authToken) .bodyString(data, ContentType.create("application/json")); return Executor.newInstance(httpClient) .execute(request) .returnResponse(); } BulkManager(final DataSiftConfig config,
                       final SimpleConsumerManager simpleConsumerManager,
                       final Backoff backoff,
                       final Metrics metrics); }
BulkManager implements Runnable { @SuppressWarnings("checkstyle:designforextension") public HttpResponse post(final String data) throws URISyntaxException, IOException { log.debug("post()"); URI uri = getUri(); String authToken = getAuthorizationToken(config); metrics.sendAttempt.mark(); log.trace("Posting to ingestion endpoint {}", data); log.debug("Posting to ingestion endpoint data length {} bytes", data.length()); Request request = Request.Post(uri) .useExpectContinue() .version(HttpVersion.HTTP_1_1) .addHeader("Auth", authToken) .bodyString(data, ContentType.create("application/json")); return Executor.newInstance(httpClient) .execute(request) .returnResponse(); } BulkManager(final DataSiftConfig config,
                       final SimpleConsumerManager simpleConsumerManager,
                       final Backoff backoff,
                       final Metrics metrics); final void setLogger(final Logger logger); final void shutdown(); final void run(); @SuppressWarnings("checkstyle:designforextension") HttpResponse post(final String data); }
BulkManager implements Runnable { @SuppressWarnings("checkstyle:designforextension") public HttpResponse post(final String data) throws URISyntaxException, IOException { log.debug("post()"); URI uri = getUri(); String authToken = getAuthorizationToken(config); metrics.sendAttempt.mark(); log.trace("Posting to ingestion endpoint {}", data); log.debug("Posting to ingestion endpoint data length {} bytes", data.length()); Request request = Request.Post(uri) .useExpectContinue() .version(HttpVersion.HTTP_1_1) .addHeader("Auth", authToken) .bodyString(data, ContentType.create("application/json")); return Executor.newInstance(httpClient) .execute(request) .returnResponse(); } BulkManager(final DataSiftConfig config,
                       final SimpleConsumerManager simpleConsumerManager,
                       final Backoff backoff,
                       final Metrics metrics); final void setLogger(final Logger logger); final void shutdown(); final void run(); @SuppressWarnings("checkstyle:designforextension") HttpResponse post(final String data);  }