Merge pull request #190 from GetFeedback/improve-abstract-http-call

Improve abstract http call
GetFeedback · Aug 25, 2022 · 471e9bf · 471e9bf
2 parents 900123e + 1262275
commit 471e9bf
Show file tree

Hide file tree

Showing 14 changed files with 152 additions and 31 deletions.
diff --git a/docs/features/http.md b/docs/features/http.md
@@ -100,3 +100,27 @@ The response can also be handled in different ways based on the status code.
 | RECORD_UPDATE        | Replace record field with the HTTP Response body | jmesPath             | Field to replace with the HTTP Response |
 | RECORD_TERMINATE     | Don't forward record to the next step            |                      |                                         |
 | RECORD_ROUTE         | Route record to specific topics                  | topics               | List of topics to route the record to   |
+
+## Configure dynamic path
+
+It's possible to configure the path dynamically using placeholders.  
+The value correspondent to the placeholder need to be available inside the record.
+
+Example:
+```yaml
+    config:
+      api: my-dummy-api
+      path: /my/path/${value.myDynamic.field}
+      topic: error
+```
+
+So record value needs to look like this:
+```json
+{
+    "myDynamic": {
+        "field": 1234
+    }
+}
+```
+
+If the value is not there, the call will fail, and the record will be routed to the error topic.
diff --git a/...toconfigure/src/main/java/dev/vox/platform/kahpp/configuration/http/AbstractHttpCall.java b/...toconfigure/src/main/java/dev/vox/platform/kahpp/configuration/http/AbstractHttpCall.java
@@ -5,15 +5,22 @@
 import dev.vox.platform.kahpp.configuration.conditional.Conditional;
 import dev.vox.platform.kahpp.configuration.http.client.ApiClient;
 import dev.vox.platform.kahpp.streams.KaHPPRecord;
+import io.burt.jmespath.jackson.JacksonRuntime;
 import io.vavr.control.Either;
 import io.vavr.control.Try;
 import java.util.Locale;
 import java.util.Map;
+import java.util.regex.Matcher;
 import javax.validation.constraints.NotBlank;
 import javax.validation.constraints.NotNull;
 import javax.validation.constraints.Pattern;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 public abstract class AbstractHttpCall implements HttpCall, Conditional {
+
+  private static final Logger LOGGER = LoggerFactory.getLogger(AbstractHttpCall.class);
+
   @NotBlank private final transient String name;
 
   /** todo: Custom validator that exposes the missing api reference */
@@ -77,6 +84,21 @@ public Either<Throwable, RecordAction> call(KaHPPRecord record) {
         .toEither();
   }
 
+  @SuppressWarnings("PMD.AvoidInstantiatingObjectsInLoops")
+  @Override
+  public Either<Throwable, RecordAction> call(KaHPPRecord record, JacksonRuntime jacksonRuntime) {
+    Matcher m = java.util.regex.Pattern.compile("\\$\\{(.*?)\\}").matcher(path);
+    while (m.find()) {
+      String value = jacksonRuntime.compile(m.group(1)).search(record.build()).textValue();
+      if (value == null) {
+        LOGGER.warn("No value found inside the record for this placeholder {}", m.group(0));
+        return Either.left(new RuntimeException("Placeholder value not found"));
+      }
+      path = path.replace(m.group(0), value);
+    }
+    return call(record);
+  }
+
   @Override
   public boolean shouldForwardRecordOnError() {
     return Boolean.parseBoolean(forwardRecordOnError);

diff --git a/...pring-autoconfigure/src/main/java/dev/vox/platform/kahpp/configuration/http/HttpCall.java b/...pring-autoconfigure/src/main/java/dev/vox/platform/kahpp/configuration/http/HttpCall.java
@@ -3,12 +3,17 @@
 import dev.vox.platform.kahpp.configuration.RecordAction;
 import dev.vox.platform.kahpp.configuration.Step;
 import dev.vox.platform.kahpp.streams.KaHPPRecord;
+import io.burt.jmespath.jackson.JacksonRuntime;
 import io.vavr.control.Either;
 
 public interface HttpCall extends Step {
   String RESPONSE_HANDLER_CONFIG = "responseHandler";
 
   Either<Throwable, RecordAction> call(KaHPPRecord record);
 
+  default Either<Throwable, RecordAction> call(KaHPPRecord record, JacksonRuntime jacksonRuntime) {
+    return call(record);
+  }
+
   boolean shouldForwardRecordOnError();
 }
diff --git a/...figure/src/main/java/dev/vox/platform/kahpp/configuration/http/HttpCallStepProcessor.java b/...figure/src/main/java/dev/vox/platform/kahpp/configuration/http/HttpCallStepProcessor.java
@@ -120,7 +120,7 @@ public HttpCallStepProcessor(
   public void process(KaHPPRecord sourceRecord) {
     final Timer.Sample sample = Timer.start(meterRegistry);
 
-    Either<Throwable, RecordAction> recordAction = step().call(sourceRecord);
+    Either<Throwable, RecordAction> recordAction = step().call(sourceRecord, jacksonRuntime);
     RecordAction action = null;
 
     if (recordAction.isLeft()) {

diff --git a/...pring-autoconfigure/src/test/java/dev/vox/platform/kahpp/integration/KaHPPMockServer.java b/...pring-autoconfigure/src/test/java/dev/vox/platform/kahpp/integration/KaHPPMockServer.java
@@ -40,22 +40,20 @@ public static void closeServer() {
     mockServer.close();
   }
 
-  public static void mockHttpResponse(String body, int statusCode) {
-    mockHttpResponse(body, statusCode, null);
+  public static void mockHttpResponse(String path, String body, int statusCode) {
+    mockHttpResponse(path, body, statusCode, null);
   }
 
-  public static void mockHttpResponse(String body, int statusCode, String responseBody) {
+  public static void mockHttpResponse(
+      String path, String body, int statusCode, String responseBody) {
     int requestTimes = REQUEST_WITHOUT_RETRIES;
     if (statusCode >= HTTP_STATUS_CODE_SERVER_ERROR) {
       requestTimes = REQUEST_WITH_RETRIES;
     }
 
     new MockServerClient("localhost", mockServer.getLocalPort())
         .when(
-            HttpRequest.request()
-                .withMethod("POST")
-                .withPath("/enrich")
-                .withBody(new JsonBody(body)),
+            HttpRequest.request().withMethod("POST").withPath(path).withBody(new JsonBody(body)),
             Times.exactly(requestTimes))
         .respond(HttpResponse.response().withStatusCode(statusCode).withBody(responseBody));
   }

diff --git a/...oconfigure/src/test/java/dev/vox/platform/kahpp/integration/http/HttpConditionalTest.java b/...oconfigure/src/test/java/dev/vox/platform/kahpp/integration/http/HttpConditionalTest.java
@@ -36,7 +36,7 @@ static void stopMockServer() {
   void shouldExecuteHttpCall() {
     Fixture fixture = loadFixture("conditional", "true");
 
-    KaHPPMockServer.mockHttpResponse(fixture.getValue(), 200, "{}");
+    KaHPPMockServer.mockHttpResponse("/enrich", fixture.getValue(), 200, "{}");
     sendFixture(TOPIC_SOURCE, fixture);
 
     ConsumerRecord<String, String> recordSink =

diff --git a/.../test/java/dev/vox/platform/kahpp/integration/http/HttpForwardAndNotProduceErrorTest.java b/.../test/java/dev/vox/platform/kahpp/integration/http/HttpForwardAndNotProduceErrorTest.java
@@ -37,7 +37,8 @@ static void stopMockServer() {
   @Test
   void successfulHttpCallShouldProduceRecordOnlyOnSinkTopic() throws JsonProcessingException {
     Fixture fixture = loadFixture("collection", "collection_6");
-    KaHPPMockServer.mockHttpResponse(fixture.getValue(), 200, "{\"new_value\":\"beautiful\"}");
+    KaHPPMockServer.mockHttpResponse(
+        "/enrich", fixture.getValue(), 200, "{\"new_value\":\"beautiful\"}");
     sendFixture(TOPIC_SOURCE, fixture);
 
     ConsumerRecord<String, String> recordSink =
@@ -55,7 +56,7 @@ void successfulHttpCallShouldProduceRecordOnlyOnSinkTopic() throws JsonProcessin
   @Test
   void erroredHttpCallShouldProduceRecordOnlyOnSinkTopic() throws JsonProcessingException {
     Fixture fixture = loadFixture("collection", "collection_6");
-    KaHPPMockServer.mockHttpResponse(fixture.getValue(), 500);
+    KaHPPMockServer.mockHttpResponse("/enrich", fixture.getValue(), 500);
     sendFixture(TOPIC_SOURCE, fixture);
 
     ConsumerRecord<String, String> recordSink =

diff --git a/...src/test/java/dev/vox/platform/kahpp/integration/http/HttpForwardAndProduceErrorTest.java b/...src/test/java/dev/vox/platform/kahpp/integration/http/HttpForwardAndProduceErrorTest.java
@@ -38,7 +38,8 @@ static void stopMockServer() {
   @Test
   void successfulHttpCallShouldProduceRecordOnlyOnSinkTopic() throws JsonProcessingException {
     Fixture fixture = loadFixture("collection", "collection_6");
-    KaHPPMockServer.mockHttpResponse(fixture.getValue(), 200, "{\"new_value\":\"beautiful\"}");
+    KaHPPMockServer.mockHttpResponse(
+        "/enrich", fixture.getValue(), 200, "{\"new_value\":\"beautiful\"}");
     sendFixture(TOPIC_SOURCE, fixture);
 
     ConsumerRecord<String, String> recordSink =
@@ -57,7 +58,7 @@ void successfulHttpCallShouldProduceRecordOnlyOnSinkTopic() throws JsonProcessin
   @SuppressWarnings("PMD.JUnitTestsShouldIncludeAssert")
   void erroredHttpCallShouldProduceRecordOnSinkAndErrorTopic() throws JsonProcessingException {
     Fixture fixture = loadFixture("collection", "collection_6");
-    KaHPPMockServer.mockHttpResponse(fixture.getValue(), 500);
+    KaHPPMockServer.mockHttpResponse("/enrich", fixture.getValue(), 500);
     sendFixture(TOPIC_SOURCE, fixture);
 
     ConsumerRecord<String, String> recordOnError =

diff --git a/...-autoconfigure/src/test/java/dev/vox/platform/kahpp/integration/http/HttpMetricsTest.java b/...-autoconfigure/src/test/java/dev/vox/platform/kahpp/integration/http/HttpMetricsTest.java
@@ -18,6 +18,7 @@
 class HttpMetricsTest extends AbstractKaHPPTest {
 
   private static final String RESPONSE_BODY_VALUE = "{}";
+  public static final String HTTP_CALL_PATH = "/enrich";
 
   @Autowired private transient MeterRegistry meterRegistry;
 
@@ -35,17 +36,17 @@ static void stopMockServer() {
   void httpMetricsWithSuccessfulTagAreCreated() {
     Fixture fixture = loadFixture("collection", "collection_6");
 
-    KaHPPMockServer.mockHttpResponse(fixture.getValue(), 200, RESPONSE_BODY_VALUE);
+    KaHPPMockServer.mockHttpResponse(HTTP_CALL_PATH, fixture.getValue(), 200, RESPONSE_BODY_VALUE);
     sendFixture(TOPIC_SOURCE, fixture);
     KafkaTestUtils.getSingleRecord(sinkTopicConsumer, TOPIC_SINK);
     assertStepMetricsCount(1.0, true);
 
-    KaHPPMockServer.mockHttpResponse(fixture.getValue(), 500);
+    KaHPPMockServer.mockHttpResponse(HTTP_CALL_PATH, fixture.getValue(), 500);
     sendFixture(TOPIC_SOURCE, fixture);
     KafkaTestUtils.getSingleRecord(sinkTopicConsumer, TOPIC_SINK);
     assertStepMetricsCount(1.0, false);
 
-    KaHPPMockServer.mockHttpResponse(fixture.getValue(), 200, RESPONSE_BODY_VALUE);
+    KaHPPMockServer.mockHttpResponse(HTTP_CALL_PATH, fixture.getValue(), 200, RESPONSE_BODY_VALUE);
     sendFixture(TOPIC_SOURCE, fixture);
     KafkaTestUtils.getSingleRecord(sinkTopicConsumer, TOPIC_SINK);
     assertStepMetricsCount(2.0, true);
@@ -56,19 +57,19 @@ void httpMetricsWithSuccessfulTagAreCreated() {
   void httpDurationMetricsWithSuccessfulTagAreCreated() {
     Fixture fixture = loadFixture("collection", "collection_6");
 
-    KaHPPMockServer.mockHttpResponse(fixture.getValue(), 200, RESPONSE_BODY_VALUE);
+    KaHPPMockServer.mockHttpResponse(HTTP_CALL_PATH, fixture.getValue(), 200, RESPONSE_BODY_VALUE);
     sendFixture(TOPIC_SOURCE, fixture);
     KafkaTestUtils.getSingleRecord(sinkTopicConsumer, TOPIC_SINK);
 
     assertTimeMetric(true, 1L);
 
-    KaHPPMockServer.mockHttpResponse(fixture.getValue(), 429, RESPONSE_BODY_VALUE);
+    KaHPPMockServer.mockHttpResponse(HTTP_CALL_PATH, fixture.getValue(), 429, RESPONSE_BODY_VALUE);
     sendFixture(TOPIC_SOURCE, fixture);
     KafkaTestUtils.getSingleRecord(errorTopicConsumer, TOPIC_ERROR);
 
     assertTimeMetric(false, 1L);
 
-    KaHPPMockServer.mockHttpResponse(fixture.getValue(), 200, RESPONSE_BODY_VALUE);
+    KaHPPMockServer.mockHttpResponse(HTTP_CALL_PATH, fixture.getValue(), 200, RESPONSE_BODY_VALUE);
     sendFixture(TOPIC_SOURCE, fixture);
     KafkaTestUtils.getSingleRecord(sinkTopicConsumer, TOPIC_SINK);
 

diff --git a/...st/java/dev/vox/platform/kahpp/integration/http/HttpNotForwardAndNotProduceErrorTest.java b/...st/java/dev/vox/platform/kahpp/integration/http/HttpNotForwardAndNotProduceErrorTest.java
@@ -37,7 +37,8 @@ static void stopMockServer() {
   @Test
   void successfulHttpCallShouldProduceRecordOnlyOnSinkTopic() throws JsonProcessingException {
     Fixture fixture = loadFixture("collection", "collection_6");
-    KaHPPMockServer.mockHttpResponse(fixture.getValue(), 200, "{\"new_value\":\"beautiful\"}");
+    KaHPPMockServer.mockHttpResponse(
+        "/enrich", fixture.getValue(), 200, "{\"new_value\":\"beautiful\"}");
     sendFixture(TOPIC_SOURCE, fixture);
 
     ConsumerRecord<String, String> recordSink =
@@ -55,7 +56,7 @@ void successfulHttpCallShouldProduceRecordOnlyOnSinkTopic() throws JsonProcessin
   @Test
   void erroredHttpCallShouldNotProduceRecord() {
     Fixture fixture = loadFixture("collection", "collection_6");
-    KaHPPMockServer.mockHttpResponse(fixture.getValue(), 500);
+    KaHPPMockServer.mockHttpResponse("/enrich", fixture.getValue(), 500);
     sendFixture(TOPIC_SOURCE, fixture);
 
     assertThatThrownBy(

diff --git a/.../test/java/dev/vox/platform/kahpp/integration/http/HttpNotForwardAndProduceErrorTest.java b/.../test/java/dev/vox/platform/kahpp/integration/http/HttpNotForwardAndProduceErrorTest.java
@@ -38,7 +38,8 @@ static void stopMockServer() {
   @Test
   void successfulHttpCallShouldProduceRecordOnlyOnSinkTopic() throws JsonProcessingException {
     Fixture fixture = loadFixture("collection", "collection_6");
-    KaHPPMockServer.mockHttpResponse(fixture.getValue(), 200, "{\"new_value\":\"beautiful\"}");
+    KaHPPMockServer.mockHttpResponse(
+        "/enrich", fixture.getValue(), 200, "{\"new_value\":\"beautiful\"}");
     sendFixture(TOPIC_SOURCE, fixture);
 
     ConsumerRecord<String, String> recordSink =
@@ -56,7 +57,7 @@ void successfulHttpCallShouldProduceRecordOnlyOnSinkTopic() throws JsonProcessin
   @Test
   void erroredHttpCallShouldProduceRecordOnlyOnErrorTopic() throws JsonProcessingException {
     Fixture fixture = loadFixture("collection", "collection_6");
-    KaHPPMockServer.mockHttpResponse(fixture.getValue(), 500);
+    KaHPPMockServer.mockHttpResponse("/enrich", fixture.getValue(), 500);
     sendFixture(TOPIC_SOURCE, fixture);
 
     ConsumerRecord<String, String> recordOnError =
@@ -76,6 +77,7 @@ void erroredHttpCallShouldProduceRecordOnlyOnErrorTopic() throws JsonProcessingE
   void htmlHttpResponseShouldGoToErrorTopic() {
     Fixture fixture = loadFixture("collection", "collection_6");
     KaHPPMockServer.mockHttpResponse(
+        "/enrich",
         fixture.getValue(),
         200,
         "\n"

diff --git a/...figure/src/test/java/dev/vox/platform/kahpp/integration/http/HttpResponseHandlerTest.java b/...figure/src/test/java/dev/vox/platform/kahpp/integration/http/HttpResponseHandlerTest.java
@@ -38,7 +38,7 @@ static void stopMockServer() {
   void shouldExecuteHttpCall() {
     Fixture fixture = loadFixture("conditional", "true");
 
-    KaHPPMockServer.mockHttpResponse(fixture.getValue(), 200, "{}");
+    KaHPPMockServer.mockHttpResponse("/enrich", fixture.getValue(), 200, "{}");
     sendFixture(TOPIC_SOURCE, fixture);
 
     ConsumerRecord<String, String> recordSink =

diff --git a/...t/java/dev/vox/platform/kahpp/unit/configuration/http/ResponseHandlerRecordRouteTest.java b/...t/java/dev/vox/platform/kahpp/unit/configuration/http/ResponseHandlerRecordRouteTest.java
@@ -73,7 +73,7 @@ void shouldRouteToTopicIfTheStatusCodeIsInRange() {
     JsonNode key = objectMapper.createObjectNode().put("key", 1);
     JsonNode value = objectMapper.createObjectNode().put("value", "foo");
 
-    KaHPPMockServer.mockHttpResponse(value.toString(), 410, "{\"foo\":\"bar\"}");
+    KaHPPMockServer.mockHttpResponse("/enrich", value.toString(), 410, "{\"foo\":\"bar\"}");
 
     Either<Throwable, RecordAction> afterCall =
         httpCall.call(KaHPPRecord.build(key, value, 1584352842123L));
@@ -92,7 +92,7 @@ void shouldNotRouteToTopicIfTheStatusCodeIsNotInRange() {
     JsonNode key = objectMapper.createObjectNode().put("key", 1);
     JsonNode value = objectMapper.createObjectNode().put("value", "foo");
 
-    KaHPPMockServer.mockHttpResponse(value.toString(), 200, "{\"foo\":\"bar\"}");
+    KaHPPMockServer.mockHttpResponse("/enrich", value.toString(), 200, "{\"foo\":\"bar\"}");
 
     Either<Throwable, RecordAction> afterCall =
         httpCall.call(KaHPPRecord.build(key, value, 1584352842123L));