checkpoint

runtime/binding-asyncapi/src/main/java/io/aklivity/zilla/runtime/binding/asyncapi/internal/config/AsyncapiBindingConfig.java

@@ -47,6 +47,7 @@
 import io.aklivity.zilla.runtime.binding.asyncapi.internal.types.String16FW;
 import io.aklivity.zilla.runtime.binding.asyncapi.internal.types.String8FW;
 import io.aklivity.zilla.runtime.binding.asyncapi.internal.types.stream.HttpBeginExFW;
+import io.aklivity.zilla.runtime.binding.asyncapi.internal.view.AsyncapiServerView;
 import io.aklivity.zilla.runtime.engine.catalog.CatalogHandler;
 import io.aklivity.zilla.runtime.engine.config.BindingConfig;
 import io.aklivity.zilla.runtime.engine.config.KindConfig;
@@ -228,33 +229,50 @@ private void attachProxyBinding(
         for (AsyncapiSchemaConfig config : configs)
         {
             Asyncapi asyncapi = config.asyncapi;
-            updateNamespace(config, composite, asyncapi);
+            //updateNamespace(config, composite, asyncapi);
         }
     }
 
     private void attachServerClientBinding(
         BindingConfig binding,
         List<AsyncapiSchemaConfig> configs)
     {
+        final Map<Integer, AsyncapiNamespaceConfig> namespaceConfigs = new HashMap<>();
         for (AsyncapiSchemaConfig config : configs)
         {
             Asyncapi asyncapi = config.asyncapi;
-            final NamespaceConfig composite = namespaceGenerator.generate(binding, asyncapi);
+            final List<AsyncapiServerView> servers =
+                namespaceGenerator.filterAsyncapiServers(asyncapi, options.asyncapis.stream()
+                    .filter(a -> a.apiLabel.equals(config.apiLabel))
+                    .flatMap(a -> a.servers.stream())
+                    .collect(Collectors.toList()));
+
+            servers.stream().collect(Collectors.groupingBy(AsyncapiServerView::getPort)).forEach((k, v) ->
+                namespaceConfigs.computeIfAbsent(k, s -> new AsyncapiNamespaceConfig()).addServersForSpec(v, config, asyncapi));
+            servers.forEach(s ->
+                s.setAsyncapiProtocol(namespaceGenerator.resolveProtocol(s.protocol(), options, asyncapi, servers)));
+        }
+
+        for (AsyncapiNamespaceConfig namespaceConfig : namespaceConfigs.values())
+        {
+            final NamespaceConfig composite = namespaceGenerator.generate(binding, namespaceConfig);
             composite.readURL = binding.readURL;
             attach.accept(composite);
-            updateNamespace(config, composite, asyncapi);
+            updateNamespace(namespaceConfig.configs, composite, namespaceConfig.asyncapis);
         }
     }
 
     private void updateNamespace(
-        AsyncapiSchemaConfig config,
+        List<AsyncapiSchemaConfig> configs,
         NamespaceConfig composite,
-        Asyncapi asyncapi)
+        List<Asyncapi> asyncapis)
     {
-        composites.put(config.schemaId, composite);
-        schemaIdsByApiId.put(config.apiLabel, config.schemaId);
-        extractChannels(asyncapi);
-        extractOperations(asyncapi);
+        //TODO: get first schemaId?
+        final int schemaId = configs.get(0).schemaId;
+        composites.put(schemaId, composite);
+        configs.forEach(c -> schemaIdsByApiId.put(c.apiLabel, schemaId));
+        asyncapis.forEach(this::extractChannels);
+        asyncapis.forEach(this::extractOperations);
     }
 
     private void extractNamespace(
@@ -402,4 +420,28 @@ private void visitAuthority(
             authority = authorityRO.wrap(value.buffer(), value.offset(), value.limit());
         }
     }
+
+    static class AsyncapiNamespaceConfig
+    {
+        List<AsyncapiServerView> servers;
+        List<Asyncapi> asyncapis;
+        List<AsyncapiSchemaConfig> configs;
+
+        AsyncapiNamespaceConfig()
+        {
+            servers = new ArrayList<>();
+            asyncapis = new ArrayList<>();
+            configs = new ArrayList<>();
+        }
+
+        private void addServersForSpec(
+            List<AsyncapiServerView> servers,
+            AsyncapiSchemaConfig config,
+            Asyncapi asyncapi)
+        {
+            this.servers.addAll(servers);
+            this.configs.add(config);
+            this.asyncapis.add(asyncapi);
+        }
+    }
 }

runtime/binding-asyncapi/src/main/java/io/aklivity/zilla/runtime/binding/asyncapi/internal/config/AsyncapiClientNamespaceGenerator.java

@@ -18,10 +18,8 @@
 import static java.util.Collections.emptyList;
 
 import java.util.List;
-import java.util.stream.Collectors;
 
 import io.aklivity.zilla.runtime.binding.asyncapi.config.AsyncapiOptionsConfig;
-import io.aklivity.zilla.runtime.binding.asyncapi.internal.model.Asyncapi;
 import io.aklivity.zilla.runtime.binding.asyncapi.internal.view.AsyncapiServerView;
 import io.aklivity.zilla.runtime.engine.config.BindingConfig;
 import io.aklivity.zilla.runtime.engine.config.MetricRefConfig;
@@ -32,22 +30,17 @@ public class AsyncapiClientNamespaceGenerator extends AsyncapiNamespaceGenerator
 {
     public NamespaceConfig generate(
         BindingConfig binding,
-        Asyncapi asyncapi)
+        AsyncapiBindingConfig.AsyncapiNamespaceConfig namespaceConfig)
     {
+        List<AsyncapiServerView> servers = namespaceConfig.servers;
         AsyncapiOptionsConfig options = binding.options != null ? (AsyncapiOptionsConfig) binding.options : EMPTY_OPTION;
         final List<MetricRefConfig> metricRefs = binding.telemetryRef != null ?
             binding.telemetryRef.metricRefs : emptyList();
 
-        this.asyncapi = asyncapi;
         this.qname = binding.qname;
         this.namespace = binding.namespace;
         this.qvault = binding.qvault;
         this.vault = binding.vault;
-        final List<AsyncapiServerView> servers =
-            filterAsyncapiServers(asyncapi.servers, options.asyncapis.stream()
-                .flatMap(a -> a.servers.stream())
-                .collect(Collectors.toList()));
-        servers.forEach(s -> s.setAsyncapiProtocol(resolveProtocol(s.protocol(), options, servers)));
 
         //TODO: keep it until we support different protocols on the same composite binding
         AsyncapiServerView serverView = servers.get(0);
@@ -58,7 +51,7 @@ public NamespaceConfig generate(
         return NamespaceConfig.builder()
                 .name(String.format("%s.%s", qname, "$composite"))
                 .inject(n -> this.injectNamespaceMetric(n, !metricRefs.isEmpty()))
-                .inject(n -> this.injectCatalog(n, asyncapi))
+                .inject(n -> this.injectCatalog(n, namespaceConfig.asyncapis))
                 .inject(n -> protocol.injectProtocolClientCache(n, metricRefs))
                 .binding()
                     .name(String.format("%s_client0", protocol.scheme))

runtime/binding-asyncapi/src/main/java/io/aklivity/zilla/runtime/binding/asyncapi/internal/config/AsyncapiNamespaceGenerator.java

@@ -65,7 +65,6 @@ public abstract class AsyncapiNamespaceGenerator
     protected static final Pattern VARIABLE = Pattern.compile("\\{([^}]*.?)\\}");
     protected final Matcher variable = VARIABLE.matcher("");
 
-    protected Asyncapi asyncapi;
     protected Map<String, Asyncapi> asyncapis;
     protected boolean isTlsEnabled;
     protected AsyncapiProtocol protocol;
@@ -76,7 +75,7 @@ public abstract class AsyncapiNamespaceGenerator
 
     public NamespaceConfig generate(
         BindingConfig binding,
-        Asyncapi asyncapi)
+        AsyncapiBindingConfig.AsyncapiNamespaceConfig namespaceConfig)
     {
         return null;
     }
@@ -92,6 +91,7 @@ public NamespaceConfig generateProxy(
     protected AsyncapiProtocol resolveProtocol(
         String protocolName,
         AsyncapiOptionsConfig options,
+        Asyncapi asyncapi,
         List<AsyncapiServerView> servers)
     {
         Pattern pattern = Pattern.compile("(http|mqtt|kafka)");
@@ -121,9 +121,10 @@ protected AsyncapiProtocol resolveProtocol(
     }
 
     protected List<AsyncapiServerView> filterAsyncapiServers(
-        Map<String, AsyncapiServer> servers,
+        Asyncapi asyncapi,
         List<AsyncapiServerConfig> serverConfigs)
     {
+        final Map<String, AsyncapiServer> servers = asyncapi.servers;
         List<AsyncapiServerView> filtered;
         Map<String, AsyncapiServerView> serverViews = servers.entrySet().stream().collect(Collectors.toMap(
             Map.Entry::getKey, e -> AsyncapiServerView.of(e.getValue(), asyncapi.components.serverVariables)));
@@ -178,26 +179,30 @@ public int[] resolvePorts(
 
     protected <C> NamespaceConfigBuilder<C> injectCatalog(
         NamespaceConfigBuilder<C> namespace,
-        Asyncapi asyncapi)
+        List<Asyncapi> asyncapis)
     {
-        if (asyncapi.components != null && asyncapi.components.schemas != null && !asyncapi.components.schemas.isEmpty())
+        for (Asyncapi asyncapi : asyncapis)
         {
-            namespace
-                .catalog()
-                    .name(INLINE_CATALOG_NAME)
-                    .type(INLINE_CATALOG_TYPE)
-                    .options(InlineOptionsConfig::builder)
-                        .subjects()
-                            .inject(this::injectSubjects)
+            if (asyncapi.components != null && asyncapi.components.schemas != null && !asyncapi.components.schemas.isEmpty())
+            {
+                namespace
+                    .catalog()
+                        .name(INLINE_CATALOG_NAME)
+                        .type(INLINE_CATALOG_TYPE)
+                        .options(InlineOptionsConfig::builder)
+                            .subjects()
+                                .inject(s -> injectSubjects(s, asyncapi))
+                                .build()
                             .build()
-                        .build()
-                    .build();
+                        .build();
+            }
         }
         return namespace;
     }
 
     protected <C> InlineSchemaConfigBuilder<C> injectSubjects(
-        InlineSchemaConfigBuilder<C> subjects)
+        InlineSchemaConfigBuilder<C> subjects,
+        Asyncapi asyncapi)
     {
         try (Jsonb jsonb = JsonbBuilder.create())
         {

runtime/binding-asyncapi/src/main/java/io/aklivity/zilla/runtime/binding/asyncapi/internal/config/AsyncapiServerNamespaceGenerator.java

@@ -18,10 +18,8 @@
 import static java.util.Collections.emptyList;
 
 import java.util.List;
-import java.util.stream.Collectors;
 
 import io.aklivity.zilla.runtime.binding.asyncapi.config.AsyncapiOptionsConfig;
-import io.aklivity.zilla.runtime.binding.asyncapi.internal.model.Asyncapi;
 import io.aklivity.zilla.runtime.binding.asyncapi.internal.view.AsyncapiServerView;
 import io.aklivity.zilla.runtime.binding.tcp.config.TcpConditionConfig;
 import io.aklivity.zilla.runtime.binding.tcp.config.TcpOptionsConfig;
@@ -36,31 +34,25 @@ public class AsyncapiServerNamespaceGenerator extends AsyncapiNamespaceGenerator
 {
     public NamespaceConfig generate(
         BindingConfig binding,
-        Asyncapi asyncapi)
+        AsyncapiBindingConfig.AsyncapiNamespaceConfig namespaceConfig)
     {
+        List<AsyncapiServerView> servers = namespaceConfig.servers;
         AsyncapiOptionsConfig options = binding.options != null ? (AsyncapiOptionsConfig) binding.options : EMPTY_OPTION;
         final List<MetricRefConfig> metricRefs = binding.telemetryRef != null ?
             binding.telemetryRef.metricRefs : emptyList();
 
-        this.asyncapi = asyncapi;
         this.qname = binding.qname;
         this.qvault = binding.qvault;
         this.namespace = binding.namespace;
 
-        final List<AsyncapiServerView> servers =
-            filterAsyncapiServers(asyncapi.servers, options.asyncapis.stream()
-                .flatMap(a -> a.servers.stream())
-                .collect(Collectors.toList()));
-        servers.forEach(s -> s.setAsyncapiProtocol(resolveProtocol(s.protocol(), options, servers)));
-
         //TODO: keep it until we support different protocols on the same composite binding
         AsyncapiServerView serverView = servers.get(0);
         this.protocol = serverView.getAsyncapiProtocol();
 
         return NamespaceConfig.builder()
                 .name(String.format("%s/%s", qname, protocol.scheme))
                 .inject(n -> this.injectNamespaceMetric(n, !metricRefs.isEmpty()))
-                .inject(n -> this.injectCatalog(n, asyncapi))
+                .inject(n -> this.injectCatalog(n, namespaceConfig.asyncapis))
                 .inject(n -> injectTcpServer(n, servers, options, metricRefs))
                 .inject(n -> injectTlsServer(n, options))
                 .binding()

runtime/binding-asyncapi/src/test/java/io/aklivity/zilla/runtime/binding/asyncapi/internal/stream/server/AsyncapiIT.java

@@ -70,4 +70,26 @@ public void shouldCreatePet() throws Exception
     {
         k3po.finish();
     }
+
+    @Test
+    @Configuration("server.multi.protocol.yaml")
+    @Specification({
+        "${mqtt}/publish.and.subscribe/client",
+        "${asyncapi}/mqtt/publish.and.subscribe/server"
+    })
+    public void shouldPublishAndSubscribeMultipleSpec() throws Exception
+    {
+        k3po.finish();
+    }
+
+    @Test
+    @Configuration("server.multi.protocol.yaml")
+    @Specification({
+        "${http}/create.pet/client",
+        "${asyncapi}/http/create.pet/server"
+    })
+    public void shouldCreatePetMultipleSpec() throws Exception
+    {
+        k3po.finish();
+    }
 }

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

@@ -20,6 +20,7 @@ catalogs:
   catalog0:
     type: test
     options:
+      id: 1
       subject: petstore
       schema: |
         asyncapi: 3.0.0