Support http to kafka proxy using openapi.yaml and asyncapi.yaml

cloud/docker-image/pom.xml

@@ -55,6 +55,12 @@
       <version>${project.version}</version>
       <scope>runtime</scope>
     </dependency>
+    <dependency>
+      <groupId>${project.groupId}</groupId>
+      <artifactId>binding-openapi-asyncapi</artifactId>
+      <version>${project.version}</version>
+      <scope>runtime</scope>
+    </dependency>
     <dependency>
       <groupId>${project.groupId}</groupId>
       <artifactId>binding-echo</artifactId>

cloud/docker-image/src/main/docker/zpm.json.template

@@ -28,6 +28,7 @@
     "io.aklivity.zilla:binding-mqtt",
     "io.aklivity.zilla:binding-mqtt-kafka",
     "io.aklivity.zilla:binding-openapi",
+    "io.aklivity.zilla:binding-openapi-asyncapi",
     "io.aklivity.zilla:binding-proxy",
     "io.aklivity.zilla:binding-sse",
     "io.aklivity.zilla:binding-sse-kafka",

incubator/binding-asyncapi.spec/src/main/scripts/io/aklivity/zilla/specs/binding/asyncapi/config/http/asyncapi.yaml

@@ -30,31 +30,34 @@ defaultContentType: application/json
 channels:
   pets:
     address: /pets
+    messages:
+      pet:
+        $ref: '#/components/messages/pet'
   showPetById:
     address: /pets/{id}
+    messages:
+      pet:
+        $ref: '#/components/messages/pet'
 
 operations:
   createPet:
     action: send
     bindings:
       http:
-        type: request
         method: POST
     channel:
       $ref: '#/channels/pets'
   listPets:
     action: receive
     bindings:
       http:
-        type: request
         method: GET
     channel:
       $ref: '#/channels/pets'
   getPets:
     action: receive
     bindings:
       http:
-        type: request
         method: GET
         query:
           type: object
@@ -68,3 +71,26 @@ components:
   correlationIds:
     petsCorrelationId:
       location: '$message.header#/idempotency-key'
+  schemas:
+    petPayload:
+      type: object
+      properties:
+        id:
+          type: integer
+          minimum: 0
+          description: Pet id.
+        name:
+          type: string
+          description: Pet name.
+        tag:
+          type: string
+          description: Tag.
+  messages:
+    pet:
+      name: Pet
+      title: Pet
+      summary: >-
+        Inform about Pet.
+      contentType: application/json
+      payload:
+        $ref: '#/components/schemas/petPayload'

incubator/binding-asyncapi.spec/src/main/scripts/io/aklivity/zilla/specs/binding/asyncapi/streams/asyncapi/http/create.pet/client.rpt

@@ -20,7 +20,7 @@ connect "zilla://streams/asyncapi0"
 
 write zilla:begin.ext ${asyncapi:beginEx()
                              .typeId(zilla:id("asyncapi"))
-                             .apiId(3833148448)
+                             .apiId(759838734)
                              .operationId("createPet")
                              .extension(http:beginEx()
                                   .typeId(zilla:id("http"))

incubator/binding-asyncapi.spec/src/main/scripts/io/aklivity/zilla/specs/binding/asyncapi/streams/asyncapi/http/create.pet/server.rpt

@@ -22,7 +22,7 @@ accepted
 
 read zilla:begin.ext ${asyncapi:matchBeginEx()
                              .typeId(zilla:id("asyncapi"))
-                             .apiId(3833148448)
+                             .apiId(759838734)
                              .operationId("createPet")
                              .extension(http:beginEx()
                                   .typeId(zilla:id("http"))

incubator/binding-asyncapi.spec/src/test/java/io/aklivity/zilla/specs/binding/asyncapi/AsyncapiFunctionsTest.java

@@ -63,6 +63,7 @@ public void shouldMatchAsyncapiBeginExtensionOnly() throws Exception
     {
         BytesMatcher matcher = AsyncapiFunctions.matchBeginEx()
             .typeId(0x00)
+            .apiId(1L)
             .extension(new byte[] {1})
             .build();
 
@@ -71,6 +72,7 @@ public void shouldMatchAsyncapiBeginExtensionOnly() throws Exception
 
         new AsyncapiBeginExFW.Builder().wrap(new UnsafeBuffer(byteBuf), 0, byteBuf.capacity())
             .typeId(0x00)
+            .apiId(1L)
             .extension(new OctetsFW.Builder().wrap(writeBuffer, 0, 1).set(new byte[] {1}).build())
             .build();
 
@@ -82,7 +84,7 @@ public void shouldMatchAsyncapiBeginExtension() throws Exception
     {
         BytesMatcher matcher = AsyncapiFunctions.matchBeginEx()
             .typeId(0x00)
-            .apiId(1)
+            .apiId(1L)
             .operationId("operationId")
             .extension(new byte[] {1})
             .build();

incubator/binding-asyncapi/src/main/java/io/aklivity/zilla/runtime/binding/asyncapi/config/AsyncapiParser.java

@@ -0,0 +1,132 @@
+/*
+ * Copyright 2021-2023 Aklivity Inc
+ *
+ * Licensed under the Aklivity Community License (the "License"); you may not use
+ * this file except in compliance with the License.  You may obtain a copy of the
+ * License at
+ *
+ *   https://www.aklivity.io/aklivity-community-license/
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OF ANY KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations under the License.
+ */
+package io.aklivity.zilla.runtime.binding.asyncapi.config;
+
+import static java.util.Collections.unmodifiableMap;
+import static org.agrona.LangUtil.rethrowUnchecked;
+
+import java.io.InputStream;
+import java.io.StringReader;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Map;
+
+import jakarta.json.Json;
+import jakarta.json.JsonObject;
+import jakarta.json.JsonReader;
+import jakarta.json.bind.Jsonb;
+import jakarta.json.bind.JsonbBuilder;
+
+import org.agrona.collections.Object2ObjectHashMap;
+import org.leadpony.justify.api.JsonSchema;
+import org.leadpony.justify.api.JsonValidationService;
+import org.leadpony.justify.api.ProblemHandler;
+
+import io.aklivity.zilla.runtime.binding.asyncapi.internal.AsyncapiBinding;
+import io.aklivity.zilla.runtime.binding.asyncapi.internal.model.Asyncapi;
+import io.aklivity.zilla.runtime.engine.config.ConfigException;
+
+public class AsyncapiParser
+{
+    private final Map<String, JsonSchema> schemas;
+
+    public AsyncapiParser()
+    {
+        Map<String, JsonSchema> schemas = new Object2ObjectHashMap<>();
+        schemas.put("2.6.0", schema("2.6.0"));
+        schemas.put("3.0.0", schema("3.0.0"));
+        this.schemas = unmodifiableMap(schemas);
+    }
+
+    public Asyncapi parse(
+        String asyncapiText)
+    {
+        Asyncapi asyncapi = null;
+
+        List<Exception> errors = new LinkedList<>();
+
+        try
+        {
+            String asyncApiVersion = detectAsyncApiVersion(asyncapiText);
+
+            JsonValidationService service = JsonValidationService.newInstance();
+            ProblemHandler handler = service.createProblemPrinter(msg -> errors.add(new ConfigException(msg)));
+            JsonSchema schema = schemas.get(asyncApiVersion);
+
+            service.createReader(new StringReader(asyncapiText), schema, handler).read();
+
+            Jsonb jsonb = JsonbBuilder.create();
+
+            asyncapi = jsonb.fromJson(asyncapiText, Asyncapi.class);
+        }
+        catch (Exception ex)
+        {
+            errors.add(ex);
+        }
+
+        if (!errors.isEmpty())
+        {
+            Exception ex = errors.remove(0);
+            errors.forEach(ex::addSuppressed);
+            rethrowUnchecked(ex);
+        }
+
+        return asyncapi;
+    }
+
+    private JsonSchema schema(
+        String version)
+    {
+        InputStream schemaInput = null;
+
+        if (version.startsWith("2.6"))
+        {
+            schemaInput =  AsyncapiBinding.class.getResourceAsStream("schema/asyncapi.2.6.schema.json");
+        }
+        else if (version.startsWith("3.0"))
+        {
+            schemaInput = AsyncapiBinding.class.getResourceAsStream("schema/asyncapi.3.0.schema.json");
+        }
+
+        JsonValidationService service = JsonValidationService.newInstance();
+
+        return service.createSchemaReaderFactoryBuilder()
+                .withSpecVersionDetection(true)
+                .build()
+                .createSchemaReader(schemaInput)
+                .read();
+    }
+
+    private String detectAsyncApiVersion(
+        String openapiText)
+    {
+        try (JsonReader reader = Json.createReader(new StringReader(openapiText)))
+        {
+            JsonObject json = reader.readObject();
+            if (json.containsKey("asyncapi"))
+            {
+                return json.getString("asyncapi");
+            }
+            else
+            {
+                throw new IllegalArgumentException("Unable to determine AsyncApi version.");
+            }
+        }
+        catch (Exception e)
+        {
+            throw new RuntimeException("Error reading AsyncApi document.", e);
+        }
+    }
+}

incubator/binding-asyncapi/src/main/java/io/aklivity/zilla/runtime/binding/asyncapi/internal/AsyncapiProxyCompositeBindingAdapter.java

@@ -127,7 +127,6 @@ public <C> BindingConfigBuilder<C> injectMqttKafkaRoutes(
                 {
                     break inject;
                 }
-
                 final AsyncapiOperation whenOperation = mqttAsyncapi.operations.get(condition.operationId);
                 final AsyncapiChannelView channel = AsyncapiChannelView.of(mqttAsyncapi.channels, whenOperation.channel);
                 final MqttKafkaConditionKind kind = whenOperation.action.equals(ASYNCAPI_SEND_ACTION_NAME) ?