@Test public void simpleWireMockTest() throws Exception { stubFor(get(urlEqualTo("/paver/v1/roads/test_road")) .willReturn(aResponse().withStatus(HttpStatus.SC_OK).withBody("{}"))); stubFor(post(urlEqualTo("/onramp/v1/roads/test_road/messages")) .willReturn(aResponse().withStatus(HttpStatus.SC_OK).withBody(createStubResponse(3, true, "accepted: ")))); String host = "localhost:" + wireMockRule.httpsPort(); OnrampOptions options = OnrampOptions.builder() .host(host) .username("user") .password("pass") .roadName("test_road") .objectMapper(mapper) .tlsConfigFactory(TLSConfig.trustAllFactory()) .build(); try (CloseableFunction<TestMessage, CompletableFuture<StandardResponse>> function = underTest.newInstance(options, 1, 1, EnqueueBehaviour.COMPLETE_EXCEPTIONALLY)) { CompletableFuture<StandardResponse> result = function.apply(new TestMessage("foo")); StandardResponse response = result.get(); assertThat(response.getTimestamp(), is(100L)); assertThat(response.isSuccess(), is(true)); assertThat(response.getMessage(), is("accepted: 0")); } verify(postRequestedFor(urlEqualTo("/onramp/v1/roads/test_road/messages")) .withRequestBody(equalTo("[{\"data\":\"foo\"}]"))); }
public CloseableFunction<T, CompletableFuture<StandardResponse>> newInstance( OnrampOptions options, int bufferSize, int maxBatchSize, @NonNull EnqueueBehaviour enqueueBehaviour) { DefinitionSupplier definitionSupplier = new DefinitionSupplier(options); PartitionPathSupplier partitionPathSupplier = new PartitionPathSupplier(definitionSupplier); MessageHashCodeFunctionSupplier hashCodeFunctionSupplier = new MessageHashCodeFunctionSupplier(partitionPathSupplier); ScheduledSupplier<Function<JsonNode, Integer>> hasherSupplier = new ScheduledSupplier<>(hashCodeFunctionSupplier, 5, MINUTES); RoadClient<JsonNode> client = new SimpleRoadClient<>(options); MessagePartitioner<JsonNode, CompletableFuture<StandardResponse>> partitioner = new MessagePartitioner<>( hasherSupplier, listFilledWith(options.getThreads(), () -> new MessageBatcher<>(bufferSize, maxBatchSize, enqueueBehaviour, client::sendMessages))); Function<T, CompletableFuture<StandardResponse>> function = partitioner.compose(options.getObjectMapper()::valueToTree); return new CloseableFunction<T, CompletableFuture<StandardResponse>>() { @Override public CompletableFuture<StandardResponse> apply(@NonNull T message) { return function.apply(message); } @Override public void close() throws Exception { partitioner.close(); hasherSupplier.close(); client.close(); } }; }
PartitionerFactory { public CloseableFunction<T, CompletableFuture<StandardResponse>> newInstance( OnrampOptions options, int bufferSize, int maxBatchSize, @NonNull EnqueueBehaviour enqueueBehaviour) { DefinitionSupplier definitionSupplier = new DefinitionSupplier(options); PartitionPathSupplier partitionPathSupplier = new PartitionPathSupplier(definitionSupplier); MessageHashCodeFunctionSupplier hashCodeFunctionSupplier = new MessageHashCodeFunctionSupplier(partitionPathSupplier); ScheduledSupplier<Function<JsonNode, Integer>> hasherSupplier = new ScheduledSupplier<>(hashCodeFunctionSupplier, 5, MINUTES); RoadClient<JsonNode> client = new SimpleRoadClient<>(options); MessagePartitioner<JsonNode, CompletableFuture<StandardResponse>> partitioner = new MessagePartitioner<>( hasherSupplier, listFilledWith(options.getThreads(), () -> new MessageBatcher<>(bufferSize, maxBatchSize, enqueueBehaviour, client::sendMessages))); Function<T, CompletableFuture<StandardResponse>> function = partitioner.compose(options.getObjectMapper()::valueToTree); return new CloseableFunction<T, CompletableFuture<StandardResponse>>() { @Override public CompletableFuture<StandardResponse> apply(@NonNull T message) { return function.apply(message); } @Override public void close() throws Exception { partitioner.close(); hasherSupplier.close(); client.close(); } }; } }
PartitionerFactory { public CloseableFunction<T, CompletableFuture<StandardResponse>> newInstance( OnrampOptions options, int bufferSize, int maxBatchSize, @NonNull EnqueueBehaviour enqueueBehaviour) { DefinitionSupplier definitionSupplier = new DefinitionSupplier(options); PartitionPathSupplier partitionPathSupplier = new PartitionPathSupplier(definitionSupplier); MessageHashCodeFunctionSupplier hashCodeFunctionSupplier = new MessageHashCodeFunctionSupplier(partitionPathSupplier); ScheduledSupplier<Function<JsonNode, Integer>> hasherSupplier = new ScheduledSupplier<>(hashCodeFunctionSupplier, 5, MINUTES); RoadClient<JsonNode> client = new SimpleRoadClient<>(options); MessagePartitioner<JsonNode, CompletableFuture<StandardResponse>> partitioner = new MessagePartitioner<>( hasherSupplier, listFilledWith(options.getThreads(), () -> new MessageBatcher<>(bufferSize, maxBatchSize, enqueueBehaviour, client::sendMessages))); Function<T, CompletableFuture<StandardResponse>> function = partitioner.compose(options.getObjectMapper()::valueToTree); return new CloseableFunction<T, CompletableFuture<StandardResponse>>() { @Override public CompletableFuture<StandardResponse> apply(@NonNull T message) { return function.apply(message); } @Override public void close() throws Exception { partitioner.close(); hasherSupplier.close(); client.close(); } }; }  }
PartitionerFactory { public CloseableFunction<T, CompletableFuture<StandardResponse>> newInstance( OnrampOptions options, int bufferSize, int maxBatchSize, @NonNull EnqueueBehaviour enqueueBehaviour) { DefinitionSupplier definitionSupplier = new DefinitionSupplier(options); PartitionPathSupplier partitionPathSupplier = new PartitionPathSupplier(definitionSupplier); MessageHashCodeFunctionSupplier hashCodeFunctionSupplier = new MessageHashCodeFunctionSupplier(partitionPathSupplier); ScheduledSupplier<Function<JsonNode, Integer>> hasherSupplier = new ScheduledSupplier<>(hashCodeFunctionSupplier, 5, MINUTES); RoadClient<JsonNode> client = new SimpleRoadClient<>(options); MessagePartitioner<JsonNode, CompletableFuture<StandardResponse>> partitioner = new MessagePartitioner<>( hasherSupplier, listFilledWith(options.getThreads(), () -> new MessageBatcher<>(bufferSize, maxBatchSize, enqueueBehaviour, client::sendMessages))); Function<T, CompletableFuture<StandardResponse>> function = partitioner.compose(options.getObjectMapper()::valueToTree); return new CloseableFunction<T, CompletableFuture<StandardResponse>>() { @Override public CompletableFuture<StandardResponse> apply(@NonNull T message) { return function.apply(message); } @Override public void close() throws Exception { partitioner.close(); hasherSupplier.close(); client.close(); } }; }  CloseableFunction<T, CompletableFuture<StandardResponse>> newInstance(
      OnrampOptions options,
      int bufferSize,
      int maxBatchSize,
      @NonNull EnqueueBehaviour enqueueBehaviour); }
PartitionerFactory { public CloseableFunction<T, CompletableFuture<StandardResponse>> newInstance( OnrampOptions options, int bufferSize, int maxBatchSize, @NonNull EnqueueBehaviour enqueueBehaviour) { DefinitionSupplier definitionSupplier = new DefinitionSupplier(options); PartitionPathSupplier partitionPathSupplier = new PartitionPathSupplier(definitionSupplier); MessageHashCodeFunctionSupplier hashCodeFunctionSupplier = new MessageHashCodeFunctionSupplier(partitionPathSupplier); ScheduledSupplier<Function<JsonNode, Integer>> hasherSupplier = new ScheduledSupplier<>(hashCodeFunctionSupplier, 5, MINUTES); RoadClient<JsonNode> client = new SimpleRoadClient<>(options); MessagePartitioner<JsonNode, CompletableFuture<StandardResponse>> partitioner = new MessagePartitioner<>( hasherSupplier, listFilledWith(options.getThreads(), () -> new MessageBatcher<>(bufferSize, maxBatchSize, enqueueBehaviour, client::sendMessages))); Function<T, CompletableFuture<StandardResponse>> function = partitioner.compose(options.getObjectMapper()::valueToTree); return new CloseableFunction<T, CompletableFuture<StandardResponse>>() { @Override public CompletableFuture<StandardResponse> apply(@NonNull T message) { return function.apply(message); } @Override public void close() throws Exception { partitioner.close(); hasherSupplier.close(); client.close(); } }; }  CloseableFunction<T, CompletableFuture<StandardResponse>> newInstance(
      OnrampOptions options,
      int bufferSize,
      int maxBatchSize,
      @NonNull EnqueueBehaviour enqueueBehaviour);  }