WIP Flexible route matching usiing vert.x instead of JAX-RS

lib/pom.xml

@@ -10,10 +10,10 @@
   </parent>
   <artifactId>kafka-hook</artifactId>
   <dependencies>
-    <dependency><!-- library uses this for the cloudevent extension, but shouldn't annotate anything jaxrs -->
-      <groupId>org.jboss.spec.javax.ws.rs</groupId>
-      <artifactId>jboss-jaxrs-api_2.1_spec</artifactId>
-      <version>${jaxrs.version}</version>
+    <dependency>
+      <groupId>io.vertx</groupId>
+      <artifactId>vertx-core</artifactId>
+      <version>${vertx.version}</version>
     </dependency>
     <dependency>
       <groupId>com.fasterxml.jackson.core</groupId>

lib/src/main/java/se/yolean/kafka/hook/ProduceFailed.java

@@ -0,0 +1,23 @@
+package se.yolean.kafka.hook;
+
+import se.yolean.kafka.hook.types.v1.HookError;
+
+
+/**
+ * Means we failed to fulfil the primary responsibility of kafka-hook.
+ */
+public class ProduceFailed extends Exception {
+
+  private static final long serialVersionUID = 1L;
+
+  private final HookError error;
+
+  public ProduceFailed(HookError error) {
+    this.error = error;
+  }
+
+  public HookError getError() {
+    return error;
+  }
+
+}

pom.xml

@@ -34,7 +34,7 @@
     <compiler-plugin.version>3.8.1</compiler-plugin.version>
     <quarkus.platform.group-id>io.quarkus</quarkus.platform.group-id>
     <!-- keep in sync with quarkus defaults -->
-    <jaxrs.version>2.0.1.Final</jaxrs.version>
+    <vertx.version>3.9.4</vertx.version>
     <jackson.version>2.11.3</jackson.version>
     <micrometer.version>1.5.6</micrometer.version>
     <junit.version>5.7.0</junit.version>

rest/pom.xml

@@ -34,11 +34,11 @@
     </dependency>
     <dependency>
       <groupId>io.quarkus</groupId>
-      <artifactId>quarkus-resteasy</artifactId>
+      <artifactId>quarkus-vertx-web</artifactId>
     </dependency>
     <dependency>
       <groupId>io.quarkus</groupId>
-      <artifactId>quarkus-resteasy-jackson</artifactId>
+      <artifactId>quarkus-jackson</artifactId>
     </dependency>
     <dependency>
       <groupId>io.quarkus</groupId>

rest/src/main/java/se/yolean/kafka/hook/rest/KafkaHookRoutes.java

@@ -0,0 +1,38 @@
+package se.yolean.kafka.hook.rest;
+
+import java.io.IOException;
+import java.io.InputStream;
+
+import javax.enterprise.context.ApplicationScoped;
+
+import io.quarkus.vertx.web.Route;
+import io.quarkus.vertx.web.Route.HandlerType;
+import io.vertx.codegen.annotations.Nullable;
+import io.vertx.core.MultiMap;
+import io.vertx.core.buffer.Buffer;
+import io.vertx.core.http.HttpMethod;
+import io.vertx.core.http.HttpServerResponse;
+import io.vertx.ext.web.RoutingContext;
+import se.yolean.kafka.hook.ProduceFailed;
+import se.yolean.kafka.hook.types.v1.HookReceipt;
+
+@ApplicationScoped
+public class KafkaHookRoutes {
+
+  @Route(methods = HttpMethod.POST, regex = ".*/v1/?(.*)", type = Route.HandlerType.BLOCKING, produces = "application/json")
+  public HookReceipt produce(RoutingContext ctx) throws IOException {
+    String type = ctx.pathParam("param0");
+    MultiMap headers = ctx.request().headers();
+    @Nullable
+    Buffer payload = ctx.getBody();
+    return hook.produce(headers, uri, "", payload);
+  }
+
+  @Route(type = HandlerType.FAILURE, produces = "application/json")
+  void unsupported(ProduceFailed e, HttpServerResponse response) {
+    response.setStatusCode(500).end(e.getError());
+  }
+
+  // TODO pixy drop-in, stuff like: "/{topic: ^[^/]+$}/messages"
+
+}

rest/src/test/java/se/yolean/kafka/hook/rest/KafkaHookResourceIntegrationTest.java

@@ -29,6 +29,7 @@
 import static org.hamcrest.CoreMatchers.hasItems;
 import static org.hamcrest.CoreMatchers.not;
 import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotEquals;
 
 import java.io.UnsupportedEncodingException;
 import java.time.Duration;
@@ -160,6 +161,44 @@ public void testProduceString() throws UnsupportedEncodingException {
     assertEquals("github.com/Yolean/kafka-hook/mytype", headers(record2).get("ce_type"));
   }
 
+  @Test
+  public void testProduceAlternativeUrls() throws UnsupportedEncodingException {
+    given()
+      .contentType(ContentType.TEXT)
+      .accept(ContentType.JSON)
+      .body("test1".getBytes())
+      .when().post("/hook/v1/mytype/with/slashes")
+      .then()
+        .body(containsString("\"offset\":" + (startOffset)))
+        .statusCode(200);
+    given()
+      .contentType(ContentType.TEXT)
+      .accept(ContentType.JSON)
+      .body("test2".getBytes())
+      .when().post("/some-prefix/v1/hook")
+      .then()
+        .body(containsString("\"offset\":" + (startOffset + 1)))
+        .statusCode(200);
+    given()
+      .contentType(ContentType.TEXT)
+      .accept(ContentType.JSON)
+      .body("test3".getBytes())
+      .when().post("/some-prefix/v1/hook/sub/type/")
+      .then()
+        .body(containsString("\"offset\":" + (startOffset + 2)))
+        .statusCode(200);
+    waitBetweenPolls();
+    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(1000));
+    Iterator<ConsumerRecord<String, String>> it = records.iterator();
+    ConsumerRecord<String, String> record1 = it.next();
+    assertEquals("github.com/Yolean/kafka-hook/mytype/with/slashes", headers(record1).get("ce_type"));
+    ConsumerRecord<String, String> record2 = it.next();
+    assertEquals("github.com/Yolean/kafka-hook/", headers(record2).get("ce_type"));
+    ConsumerRecord<String, String> record3 = it.next();
+    assertNotEquals("github.com/Yolean/kafka-hook/sub/type/", headers(record3).get("ce_type"));
+    assertEquals("github.com/Yolean/kafka-hook/sub/type", headers(record3).get("ce_type"));
+  }
+
   @Test
   public void testCloudeventsDistributedTracingExtensionWithEnvoyHeaders() {
     given()