Http-Kafka AsyncAPI (#1072)

aklivity · Jun 7, 2024 · f7b1423 · f7b1423
1 parent 6f96b91
commit f7b1423
Show file tree

Hide file tree

Showing 27 changed files with 1,866 additions and 92 deletions.
diff --git a/runtime/binding-asyncapi/pom.xml b/runtime/binding-asyncapi/pom.xml
@@ -79,6 +79,12 @@
       <version>${project.version}</version>
       <scope>provided</scope>
     </dependency>
+    <dependency>
+      <groupId>io.aklivity.zilla</groupId>
+      <artifactId>binding-http-kafka</artifactId>
+      <version>${project.version}</version>
+      <scope>provided</scope>
+    </dependency>
     <dependency>
       <groupId>io.aklivity.zilla</groupId>
       <artifactId>binding-tls</artifactId>
@@ -157,6 +163,13 @@
       <version>${project.version}</version>
       <scope>test</scope>
     </dependency>
+    <dependency>
+      <groupId>io.aklivity.zilla</groupId>
+      <artifactId>binding-http-kafka</artifactId>
+      <type>test-jar</type>
+      <version>${project.version}</version>
+      <scope>test</scope>
+    </dependency>
     <dependency>
       <groupId>org.openjdk.jmh</groupId>
       <artifactId>jmh-core</artifactId>

diff --git a/...ava/io/aklivity/zilla/runtime/binding/asyncapi/internal/config/AsyncapiBindingConfig.java b/...ava/io/aklivity/zilla/runtime/binding/asyncapi/internal/config/AsyncapiBindingConfig.java
@@ -199,7 +199,8 @@ public void attach(
             NamespaceConfig v = entry.getValue();
             List<BindingConfig> bindings = v.bindings.stream()
                 .filter(b -> b.type.equals("mqtt") || b.type.equals("http") ||
-                    b.type.equals("kafka") && b.kind == CACHE_CLIENT || b.type.equals("mqtt-kafka"))
+                    b.type.equals("kafka") && b.kind == CACHE_CLIENT || b.type.equals("mqtt-kafka") ||
+                    b.type.equals("http-kafka"))
                 .collect(toList());
             extractResolveId(k, bindings);
             extractNamespace(k, bindings);
@@ -224,7 +225,8 @@ private void attachProxyBinding(
                         Object2ObjectHashMap::new));
 
         namespaceGenerator.init(binding);
-        final NamespaceConfig composite = namespaceGenerator.generateProxy(binding, asyncapis, schemaIdsByApiId::get);
+        final List<String> labels = configs.stream().map(c -> c.apiLabel).collect(toList());
+        final NamespaceConfig composite = namespaceGenerator.generateProxy(binding, asyncapis, schemaIdsByApiId::get, labels);
         composite.readURL = binding.readURL;
         attach.accept(composite);
         updateNamespace(configs, composite, new ArrayList<>(asyncapis.values()));

diff --git a/...livity/zilla/runtime/binding/asyncapi/internal/config/AsyncapiConditionConfigAdapter.java b/...livity/zilla/runtime/binding/asyncapi/internal/config/AsyncapiConditionConfigAdapter.java
@@ -42,17 +42,26 @@ public JsonObject adaptToJson(
         JsonObjectBuilder object = Json.createObjectBuilder();
 
         object.add(API_ID_NAME, asyncapiCondition.apiId);
-        object.add(OPERATION_ID_NAME, asyncapiCondition.operationId);
 
+        if (asyncapiCondition.operationId != null)
+        {
+            object.add(OPERATION_ID_NAME, asyncapiCondition.operationId);
+        }
         return object.build();
     }
 
     @Override
     public ConditionConfig adaptFromJson(
         JsonObject object)
     {
-        String apiId = object.getString(API_ID_NAME);
-        String operationId = object.getString(OPERATION_ID_NAME);
+        String apiId = object.containsKey(API_ID_NAME)
+            ? object.getString(API_ID_NAME)
+            : null;
+
+        String operationId = object.containsKey(OPERATION_ID_NAME)
+            ? object.getString(OPERATION_ID_NAME)
+            : null;
+
         return new AsyncapiConditionConfig(apiId, operationId);
     }
 }