Support specific server in AsyncAPI spec in asyncapi binding

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

@@ -22,7 +22,7 @@ info:
   version: 1.0.0
 servers:
   plain:
-    host: localhost:8080
+    host: localhost:8088
     protocol: http
     protocolVersion: '2.0'
 defaultContentType: application/json

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

@@ -27,6 +27,11 @@ bindings:
     options:
       specs:
         http_api: http/asyncapi.yaml
+      server: plain
+      tcp:
+        host: localhost
+        port:
+          - 8080
       tls:
         keys:
           - localhost

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

@@ -23,4 +23,9 @@ bindings:
     options:
       specs:
         http_api: http/asyncapi.yaml
+      server: plain
+      tcp:
+        host: localhost
+        port:
+          - 8080
     exit: asyncapi0

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

@@ -23,4 +23,5 @@ bindings:
     options:
       specs:
         mqtt_api: mqtt/asyncapi.yaml
+      server: plain
     exit: asyncapi0

incubator/binding-asyncapi.spec/src/main/scripts/io/aklivity/zilla/specs/binding/asyncapi/schema/asyncapi.schema.patch.json

@@ -52,6 +52,11 @@
                                     "maxProperties": 1
                                 }
                             },
+                            "server":
+                            {
+                                "title": "AsyncAPI Server",
+                                "type": "string"
+                            },
                             "tcp":
                             {
                                 "$ref": "#/$defs/options/binding/tcp"

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(2560287392)
+                             .apiId(3999379148)
                              .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(2560287392)
+                             .apiId(3999379148)
                              .operationId("createPet")
                              .extension(http:beginEx()
                                   .typeId(zilla:id("http"))

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

@@ -26,6 +26,7 @@
 public final class AsyncapiOptionsConfig extends OptionsConfig
 {
     public final List<AsyncapiConfig> specs;
+    public final String server;
     public final TcpOptionsConfig tcp;
     public final TlsOptionsConfig tls;
     public final HttpOptionsConfig http;
@@ -45,13 +46,15 @@ public static <T> AsyncapiOptionsConfigBuilder<T> builder(
 
     public AsyncapiOptionsConfig(
         List<AsyncapiConfig> specs,
+        String server,
         TcpOptionsConfig tcp,
         TlsOptionsConfig tls,
         HttpOptionsConfig http,
         KafkaOptionsConfig kafka,
         AsyncapiMqttKafkaConfig mqttKafka)
     {
         this.specs = specs;
+        this.server = server;
         this.http = http;
         this.tcp = tcp;
         this.tls = tls;

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

@@ -41,6 +41,7 @@ public final class AsyncapiOptionsConfigBuilder<T> extends ConfigBuilder<T, Asyn
     private final Function<OptionsConfig, T> mapper;
 
     public List<AsyncapiConfig> specs;
+    private String server;
     private TcpOptionsConfig tcp;
     private TlsOptionsConfig tls;
     private HttpOptionsConfig http;
@@ -67,6 +68,13 @@ public AsyncapiOptionsConfigBuilder<T> specs(
         return this;
     }
 
+    public AsyncapiOptionsConfigBuilder<T> server(
+        String server)
+    {
+        this.server = server;
+        return this;
+    }
+
     public AsyncapiOptionsConfigBuilder<T> tcp(
         TcpOptionsConfig tcp)
     {
@@ -105,6 +113,6 @@ public AsyncapiOptionsConfigBuilder<T> mqttKafka(
     @Override
     public T build()
     {
-        return mapper.apply(new AsyncapiOptionsConfig(specs, tcp, tls, http, kafka, mqttKafka));
+        return mapper.apply(new AsyncapiOptionsConfig(specs, server, tcp, tls, http, kafka, mqttKafka));
     }
 }

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

@@ -21,6 +21,7 @@
 
 import io.aklivity.zilla.runtime.binding.asyncapi.config.AsyncapiConfig;
 import io.aklivity.zilla.runtime.binding.asyncapi.config.AsyncapiOptionsConfig;
+import io.aklivity.zilla.runtime.binding.asyncapi.internal.model.AsyncapiServer;
 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.CompositeBindingAdapterSpi;
@@ -47,12 +48,14 @@ public BindingConfig adapt(
             binding.telemetryRef.metricRefs : emptyList();
 
         //TODO: add composite for all servers
-        AsyncapiServerView firstServer = AsyncapiServerView.of(asyncapi.servers.entrySet().iterator().next().getValue());
+        final AsyncapiServer server = asyncapi.servers
+            .getOrDefault(options.server, asyncapi.servers.entrySet().iterator().next().getValue());
+        AsyncapiServerView serverView = AsyncapiServerView.of(server);
         this.qname = binding.qname;
         this.namespace = binding.namespace;
         this.qvault = binding.qvault;
         this.vault = binding.vault;
-        this.protocol = resolveProtocol(firstServer.protocol(), options);
+        this.protocol = resolveProtocol(serverView.protocol(), options);
         this.isTlsEnabled = protocol.isSecure();
 
         return BindingConfig.builder(binding)

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

@@ -25,7 +25,6 @@
 
 import io.aklivity.zilla.runtime.binding.asyncapi.internal.model.Asyncapi;
 import io.aklivity.zilla.runtime.binding.asyncapi.internal.model.AsyncapiMessage;
-import io.aklivity.zilla.runtime.binding.asyncapi.internal.model.AsyncapiServer;
 import io.aklivity.zilla.runtime.binding.asyncapi.internal.view.AsyncapiMessageView;
 import io.aklivity.zilla.runtime.binding.asyncapi.internal.view.AsyncapiServerView;
 import io.aklivity.zilla.runtime.engine.config.BindingConfigBuilder;
@@ -120,20 +119,6 @@ protected boolean hasJsonContentType()
 
     protected abstract boolean isSecure();
 
-    protected int[] resolvePorts()
-    {
-        requireNonNull(scheme);
-        int[] ports = null;
-
-        for (AsyncapiServer s : asyncApi.servers.values())
-        {
-            String[] hostAndPort = s.host.split(":");
-            ports = new int[] {Integer.parseInt(hostAndPort[1])};
-            break;
-        }
-        return ports;
-    }
-
     protected URI findFirstServerUrlWithScheme(
         String scheme)
     {

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

@@ -17,10 +17,12 @@
 import static io.aklivity.zilla.runtime.engine.config.KindConfig.SERVER;
 import static java.util.Collections.emptyList;
 
+import java.util.Collections;
 import java.util.List;
 
 import io.aklivity.zilla.runtime.binding.asyncapi.config.AsyncapiConfig;
 import io.aklivity.zilla.runtime.binding.asyncapi.config.AsyncapiOptionsConfig;
+import io.aklivity.zilla.runtime.binding.asyncapi.internal.model.AsyncapiServer;
 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;
@@ -33,6 +35,7 @@
 
 public class AsyncapiServerCompositeBindingAdapter extends AsyncapiCompositeBindingAdapter implements CompositeBindingAdapterSpi
 {
+    private List<String> sni;
     private int[] compositePorts;
 
     @Override
@@ -52,12 +55,16 @@ public BindingConfig adapt(
             binding.telemetryRef.metricRefs : emptyList();
 
         //TODO: add composite for all servers
-        AsyncapiServerView firstServer = AsyncapiServerView.of(asyncapi.servers.entrySet().iterator().next().getValue());
+        final AsyncapiServer server = asyncapi.servers
+            .getOrDefault(options.server, asyncapi.servers.entrySet().iterator().next().getValue());
+        AsyncapiServerView serverView = AsyncapiServerView.of(server);
+        final String[] hostAndPort = server.host.split(":");
 
         this.qname = binding.qname;
         this.qvault = binding.qvault;
-        this.protocol = resolveProtocol(firstServer.protocol(), options);
-        this.compositePorts = protocol.resolvePorts();
+        this.protocol = resolveProtocol(serverView.protocol(), options);
+        this.sni = options.tls != null ? options.tls.sni : Collections.singletonList(hostAndPort[0]);
+        this.compositePorts = options.tcp != null ? options.tcp.ports : new int[] {Integer.parseInt(hostAndPort[1])};
         this.isTlsEnabled = protocol.isSecure();
 
         return BindingConfig.builder(binding)

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

@@ -51,6 +51,7 @@
 public final class AsyncapiOptionsConfigAdapter implements OptionsConfigAdapterSpi, JsonbAdapter<OptionsConfig, JsonObject>
 {
     private static final String SPECS_NAME = "specs";
+    private static final String SERVER_NAME = "server";
     private static final String TCP_NAME = "tcp";
     private static final String TLS_NAME = "tls";
     private static final String HTTP_NAME = "http";
@@ -102,6 +103,11 @@ public JsonObject adaptToJson(
             object.add(SPECS_NAME, specs);
         }
 
+        if (asyncapiOptions.server != null)
+        {
+            object.add(SERVER_NAME, asyncapiOptions.server);
+        }
+
         if (asyncapiOptions.tcp != null)
         {
             final TcpOptionsConfig tcp = asyncapiOptions.tcp;
@@ -171,6 +177,11 @@ public OptionsConfig adaptFromJson(
             : null;
         asyncapiOptions.specs(specs);
 
+        if (object.containsKey(SERVER_NAME))
+        {
+            asyncapiOptions.server(object.getString(SERVER_NAME));
+        }
+
         if (object.containsKey(TCP_NAME))
         {
             final JsonObject tcp = object.getJsonObject(TCP_NAME);

incubator/binding-asyncapi/src/test/java/io/aklivity/zilla/runtime/binding/asyncapi/internal/config/AsyncapiOptionsConfigAdapterTest.java

@@ -88,6 +88,7 @@ public void shouldReadOptionsMqtt() throws IOException
                     "{" +
                         "\"mqtt-api\":\"mqtt/asyncapi.yaml\"," +
                     "}," +
+                    "\"server\":\"plain\"," +
                     "\"tcp\":" +
                     "{" +
                         "\"host\":\"localhost\"," +
@@ -121,6 +122,7 @@ public void shouldReadOptionsMqtt() throws IOException
         AsyncapiConfig asyncapi = options.specs.get(0);
         assertThat(asyncapi.location, equalTo("mqtt/asyncapi.yaml"));
         assertThat(asyncapi.asyncapi, instanceOf(Asyncapi.class));
+        assertThat(options.server, equalTo("plain"));
         assertThat(options.tcp.host, equalTo("localhost"));
         assertThat(options.tcp.ports, equalTo(new int[] { 7183 }));
         assertThat(options.tls.keys, equalTo(asList("localhost")));
@@ -141,6 +143,7 @@ public void shouldWriteOptionsMqtt() throws IOException
         AsyncapiOptionsConfig options = AsyncapiOptionsConfig.builder()
             .inject(Function.identity())
             .specs(specs)
+            .server("plain")
             .tcp(TcpOptionsConfig.builder()
                 .host("localhost")
                 .ports(new int[] { 7183 })
@@ -170,6 +173,7 @@ public void shouldWriteOptionsMqtt() throws IOException
                 "{" +
                     "\"mqtt-api\":\"mqtt/asyncapi.yaml\"" +
                 "}," +
+                "\"server\":\"plain\"," +
                 "\"tcp\":" +
                 "{" +
                     "\"host\":\"localhost\"," +