SSE asyncapi server, client

runtime/binding-asyncapi/pom.xml

@@ -61,6 +61,12 @@
       <version>${project.version}</version>
       <scope>provided</scope>
     </dependency>
+    <dependency>
+      <groupId>io.aklivity.zilla</groupId>
+      <artifactId>binding-sse</artifactId>
+      <version>${project.version}</version>
+      <scope>provided</scope>
+    </dependency>
     <dependency>
       <groupId>io.aklivity.zilla</groupId>
       <artifactId>binding-kafka</artifactId>
@@ -189,7 +195,7 @@
         <artifactId>flyweight-maven-plugin</artifactId>
         <version>${project.version}</version>
         <configuration>
-          <scopeNames>core mqtt http kafka asyncapi</scopeNames>
+          <scopeNames>core mqtt http sse kafka asyncapi</scopeNames>
           <packageName>io.aklivity.zilla.runtime.binding.asyncapi.internal.types</packageName>
         </configuration>
         <executions>
@@ -224,11 +230,13 @@
               <includes>io/aklivity/zilla/specs/binding/asyncapi/schema/asyncapi.schema.patch.json,
                 io/aklivity/zilla/specs/binding/mqtt/kafka/schema/mqtt.kafka.schema.patch.json,
                 io/aklivity/zilla/specs/binding/http/schema/http.schema.patch.json,
+                io/aklivity/zilla/specs/binding/sse/schema/sse.schema.patch.json,
                 io/aklivity/zilla/specs/binding/kafka/schema/kafka.schema.patch.json,
                 io/aklivity/zilla/specs/binding/tcp/schema/tcp.schema.patch.json,
                 io/aklivity/zilla/specs/binding/tls/schema/tls.schema.patch.json,
                 io/aklivity/zilla/specs/binding/asyncapi/schema/asyncapi.2.6.0.schema.json,
-                io/aklivity/zilla/specs/binding/asyncapi/schema/asyncapi.3.0.1.schema.json
+                io/aklivity/zilla/specs/binding/asyncapi/schema/asyncapi.3.0.1.schema.json,
+                io/aklivity/zilla/specs/binding/asyncapi/schema/asyncapi.3.0.2_zilla.schema.json
               </includes>
               <outputDirectory>${project.build.directory}/classes</outputDirectory>
             </configuration>

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

@@ -51,6 +51,7 @@ public AsyncapiParser()
         Map<String, JsonSchema> schemas = new Object2ObjectHashMap<>();
         schemas.put("2.6", schema("2.6.0"));
         schemas.put("3.0", schema("3.0.1"));
+        schemas.put("3.0.2", schema("3.0.2_zilla"));
         this.schemas = unmodifiableMap(schemas);
     }
 

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

@@ -16,6 +16,7 @@
 
 import static io.aklivity.zilla.runtime.engine.config.KindConfig.CACHE_CLIENT;
 import static io.aklivity.zilla.runtime.engine.config.KindConfig.PROXY;
+import static io.aklivity.zilla.runtime.engine.config.KindConfig.SERVER;
 import static java.util.stream.Collectors.toList;
 
 import java.util.ArrayList;
@@ -197,11 +198,32 @@ public void attach(
         {
             Integer k = entry.getKey();
             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("http-kafka"))
-                .collect(toList());
+            List<BindingConfig> bindings;
+            boolean containsSse = v.bindings.stream().anyMatch(b -> b.type.equals("sse"));
+            if (containsSse)
+            {
+                if (binding.kind.equals(SERVER))
+                {
+                    bindings = v.bindings.stream()
+                        .filter(b -> b.type.equals("http") || b.type.equals("http-kafka") || b.type.equals("sse"))
+                        .collect(toList());
+                }
+                else
+                {
+                    bindings = v.bindings.stream()
+                        .filter(b -> b.type.equals("sse"))
+                        .collect(toList());
+                }
+            }
+            else
+            {
+                bindings = v.bindings.stream()
+                    .filter(b -> b.type.equals("mqtt") || b.type.equals("http") || b.type.equals("sse") ||
+                        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);
         }

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

@@ -32,14 +32,11 @@ public NamespaceConfig generate(
         BindingConfig binding,
         AsyncapiNamespaceConfig namespaceConfig)
     {
-        List<AsyncapiServerView> servers = namespaceConfig.servers;
-        AsyncapiOptionsConfig options = binding.options != null ? (AsyncapiOptionsConfig) binding.options : EMPTY_OPTION;
+        final List<AsyncapiServerView> servers = namespaceConfig.servers;
+        final AsyncapiOptionsConfig options = binding.options != null ? (AsyncapiOptionsConfig) binding.options : EMPTY_OPTION;
         final List<MetricRefConfig> metricRefs = binding.telemetryRef != null ?
             binding.telemetryRef.metricRefs : emptyList();
 
-        //TODO: keep it until we support different protocols on the same composite binding
-        AsyncapiServerView serverView = servers.get(0);
-        this.protocol = serverView.getAsyncapiProtocol();
         int[] compositeSecurePorts = resolvePorts(servers, true);
         this.isTlsEnabled =  compositeSecurePorts.length > 0;
 
@@ -48,24 +45,70 @@ public NamespaceConfig generate(
                 .name(String.format("%s.%s-%s", qname, "$composite", namespace))
                 .inject(n -> this.injectNamespaceMetric(n, !metricRefs.isEmpty()))
                 .inject(n -> this.injectCatalog(n, namespaceConfig.asyncapis))
-                .inject(n -> protocol.injectProtocolClientCache(n, metricRefs))
-                .binding()
-                    .name(String.format("%s_client0", protocol.scheme))
-                    .type(protocol.scheme)
-                    .kind(CLIENT)
-                    .inject(b -> this.injectMetrics(b, metricRefs, protocol.scheme))
-                    .inject(protocol::injectProtocolClientOptions)
-                    .exit(isTlsEnabled ? "tls_client0" : "tcp_client0")
-                    .build()
-                .inject(n -> injectTlsClient(n, options, metricRefs))
+                .inject(n -> this.injectProtocolClients(n, servers, metricRefs))
+                .inject(n -> this.injectProtocolRelatedBindings(n, servers, metricRefs))
+                .inject(n -> this.injectTlsClient(n, options, metricRefs))
+                .inject(n -> this.injectTcpClient(n, servers, options, metricRefs))
+                .build();
+    }
+
+    private <C> NamespaceConfigBuilder<C> injectTcpClient(
+        NamespaceConfigBuilder<C> namespace,
+        List<AsyncapiServerView> servers,
+        AsyncapiOptionsConfig options,
+        List<MetricRefConfig> metricRefs)
+    {
+        for (AsyncapiServerView server : servers)
+        {
+            final AsyncapiProtocol protocol = server.getAsyncapiProtocol();
+            namespace = namespace
                 .binding()
                     .name("tcp_client0")
                     .type("tcp")
                     .kind(CLIENT)
-                    .inject(b -> this.injectMetrics(b, metricRefs, "tcp"))
+                    .inject(b -> this.injectMetrics(b, metricRefs))
                     .options(!protocol.scheme.equals(AyncapiKafkaProtocol.SCHEME) ? options.tcp : null)
-                    .build()
                 .build();
+        }
+
+        return namespace;
+    }
+
+    private <C> NamespaceConfigBuilder<C> injectProtocolClients(
+        NamespaceConfigBuilder<C> namespace,
+        List<AsyncapiServerView> servers,
+        List<MetricRefConfig> metricRefs)
+    {
+        for (AsyncapiServerView server : servers)
+        {
+            final AsyncapiProtocol protocol = server.getAsyncapiProtocol();
+            final String scheme = protocol.scheme;
+            final String exit = "sse".equals(scheme) ? "http_client0" : isTlsEnabled ? "tls_client0" : "tcp_client0";
+            namespace = namespace
+                .inject(n -> protocol.injectProtocolClientCache(n, metricRefs))
+                .binding()
+                    .name(String.format("%s_client0", scheme))
+                    .type(scheme)
+                    .kind(CLIENT)
+                    .inject(b -> this.injectMetrics(b, metricRefs))
+                    .inject(protocol::injectProtocolClientOptions)
+                    .exit(exit)
+                .build();
+        }
+        return  namespace;
+    }
+
+    protected <C> NamespaceConfigBuilder<C> injectProtocolRelatedBindings(
+        NamespaceConfigBuilder<C> namespace,
+        List<AsyncapiServerView> servers,
+        List<MetricRefConfig> metricRefs)
+    {
+        for (AsyncapiServerView server : servers)
+        {
+            final AsyncapiProtocol protocol = server.getAsyncapiProtocol();
+            namespace = protocol.injectProtocolRelatedClientBindings(namespace, metricRefs, isTlsEnabled);
+        }
+        return namespace;
     }
 
     private <C> NamespaceConfigBuilder<C> injectTlsClient(
@@ -80,7 +123,7 @@ private <C> NamespaceConfigBuilder<C> injectTlsClient(
                     .name("tls_client0")
                     .type("tls")
                     .kind(CLIENT)
-                    .inject(b -> this.injectMetrics(b, metricRefs, "tls"))
+                    .inject(b -> this.injectMetrics(b, metricRefs))
                     .options(options.tls)
                     .vault(String.format("%s:%s", this.namespace, vault))
                     .exit("tcp_client0")

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

@@ -107,21 +107,40 @@ public <C> BindingConfigBuilder<C> injectProtocolServerRoutes(
             for (Map.Entry<String, AsyncapiServer> entry : asyncapi.servers.entrySet())
             {
                 AsyncapiServerView server = AsyncapiServerView.of(entry.getValue());
-                for (String name : asyncapi.operations.keySet())
+                if ("http".equals(server.protocol()))
                 {
-                    AsyncapiOperation operation = asyncapi.operations.get(name);
-                    AsyncapiChannelView channel = AsyncapiChannelView.of(asyncapi.channels, operation.channel);
-                    String path = channel.address().replaceAll("\\{[^}]+\\}", "*");
-                    String method = operation.bindings.get("http").method;
-                    binding
-                        .route()
-                        .exit(qname)
-                        .when(HttpConditionConfig::builder)
-                            .header(":path", path)
-                            .header(":method", method)
-                            .build()
-                        .inject(route -> injectHttpServerRouteGuarded(route, server))
-                        .build();
+                    for (String name : asyncapi.operations.keySet())
+                    {
+                        AsyncapiOperation operation = asyncapi.operations.get(name);
+                        AsyncapiChannelView channel = AsyncapiChannelView.of(asyncapi.channels, operation.channel);
+                        String path = channel.address().replaceAll("\\{[^}]+\\}", "*");
+                        String method = operation.bindings.get("http").method;
+                        binding
+                            .route()
+                            .exit(qname)
+                            .when(HttpConditionConfig::builder)
+                                .header(":path", path)
+                                .header(":method", method)
+                                .build()
+                            .inject(route -> injectHttpServerRouteGuarded(route, server))
+                            .build();
+                    }
+                }
+                else if ("sse".equals(server.protocol()))
+                {
+                    for (String name : asyncapi.operations.keySet())
+                    {
+                        AsyncapiOperation operation = asyncapi.operations.get(name);
+                        AsyncapiChannelView channel = AsyncapiChannelView.of(asyncapi.channels, operation.channel);
+                        String path = channel.address().replaceAll("\\{[^}]+\\}", "*");
+                        binding
+                            .route()
+                            .exit("sse_server0")
+                            .when(HttpConditionConfig::builder)
+                                .header(":path", path)
+                                .build()
+                            .build();
+                    }
                 }
             }
         }
@@ -154,10 +173,11 @@ private <C> HttpOptionsConfigBuilder<C> injectHttpServerRequests(
                 AsyncapiOperation operation = asyncapi.operations.get(name);
                 AsyncapiChannelView channel = AsyncapiChannelView.of(asyncapi.channels, operation.channel);
                 String path = channel.address();
-                Method method = Method.valueOf(operation.bindings.get("http").method);
-                if (channel.messages() != null && !channel.messages().isEmpty() ||
+
+                if (operation.bindings != null && channel.messages() != null && !channel.messages().isEmpty() ||
                     channel.parameters() != null && !channel.parameters().isEmpty())
                 {
+                    Method method = Method.valueOf(operation.bindings.get("http").method);
                     options
                         .request()
                             .path(path)

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

@@ -16,6 +16,7 @@
 
 import static com.fasterxml.jackson.dataformat.yaml.YAMLGenerator.Feature.MINIMIZE_QUOTES;
 import static com.fasterxml.jackson.dataformat.yaml.YAMLGenerator.Feature.WRITE_DOC_START_MARKER;
+import static io.aklivity.zilla.runtime.common.feature.FeatureFilter.INCUBATOR_ENABLED;
 import static java.util.stream.Collectors.toList;
 import static org.agrona.LangUtil.rethrowUnchecked;
 
@@ -67,7 +68,6 @@ public abstract class AsyncapiNamespaceGenerator
 
     protected Map<String, Asyncapi> asyncapis;
     protected boolean isTlsEnabled;
-    protected AsyncapiProtocol protocol;
     protected String qname;
     protected String namespace;
     protected String qvault;
@@ -104,7 +104,7 @@ protected AsyncapiProtocol resolveProtocol(
         List<Asyncapi> asyncapis,
         List<AsyncapiServerView> servers)
     {
-        Pattern pattern = Pattern.compile("(http|mqtt|kafka)");
+        Pattern pattern = Pattern.compile("(http|sse|mqtt|kafka)");
         Matcher matcher = pattern.matcher(protocolName);
         AsyncapiProtocol protocol = null;
         if (matcher.find())
@@ -114,6 +114,14 @@ protected AsyncapiProtocol resolveProtocol(
             case "http":
                 protocol = new AsyncapiHttpProtocol(qname, asyncapis, options, protocolName);
                 break;
+            case "sse":
+            case "secure-sse":
+                if (INCUBATOR_ENABLED)
+                {
+                    final boolean httpServerAvailable = servers.stream().anyMatch(s -> "http".equals(s.protocol()));
+                    protocol = new AsyncapiSseProtocol(qname, httpServerAvailable, asyncapis, options, protocolName);
+                }
+                break;
             case "mqtt":
                 protocol = new AsyncapiMqttProtocol(qname, asyncapis, options, protocolName, namespace);
                 break;
@@ -187,6 +195,20 @@ public int[] resolvePorts(
         return ports;
     }
 
+    public int[] resolvePortForServer(
+        AsyncapiServerView server,
+        boolean secure)
+    {
+        int[] ports = {};
+
+        if (server.getAsyncapiProtocol().isSecure() == secure)
+        {
+            ports = new int[] { server.getPort() };
+        }
+
+        return ports;
+    }
+
     protected <C> NamespaceConfigBuilder<C> injectCatalog(
         NamespaceConfigBuilder<C> namespace,
         List<Asyncapi> asyncapis)
@@ -276,8 +298,7 @@ protected static String writeSchemaYaml(
 
     protected  <C> BindingConfigBuilder<C> injectMetrics(
         BindingConfigBuilder<C> binding,
-        List<MetricRefConfig> metricRefs,
-        String protocol)
+        List<MetricRefConfig> metricRefs)
     {
         List<MetricRefConfig> metrics = metricRefs.stream()
             .filter(m -> m.name.startsWith("stream."))

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

@@ -45,22 +45,49 @@ public abstract class AsyncapiProtocol
     protected Map<String, String> securitySchemes;
     protected boolean isJwtEnabled;
     public final String scheme;
+    public final String tcpRoute;
     public final String protocol;
 
     protected AsyncapiProtocol(
         String qname,
         List<Asyncapi> asyncapis,
         String protocol,
         String scheme)
+    {
+        this(qname, asyncapis, protocol, scheme, scheme);
+    }
+
+    protected AsyncapiProtocol(
+        String qname,
+        List<Asyncapi> asyncapis,
+        String protocol,
+        String scheme,
+        String tcpRoute)
     {
         this.qname = qname;
         this.asyncapis = asyncapis;
         this.protocol = protocol;
         this.scheme = scheme;
+        this.tcpRoute = tcpRoute;
         this.securitySchemes = resolveSecuritySchemes();
         this.isJwtEnabled = !securitySchemes.isEmpty();
     }
 
+    public <C> NamespaceConfigBuilder<C> injectProtocolRelatedServerBindings(
+        NamespaceConfigBuilder<C> namespace,
+        List<MetricRefConfig> metricRefs)
+    {
+        return namespace;
+    }
+
+    public <C> NamespaceConfigBuilder<C> injectProtocolRelatedClientBindings(
+        NamespaceConfigBuilder<C> namespace,
+        List<MetricRefConfig> metricRefs,
+        boolean isTlsEnabled)
+    {
+        return namespace;
+    }
+
     public abstract <C>BindingConfigBuilder<C> injectProtocolServerOptions(
         BindingConfigBuilder<C> binding);
 
@@ -153,8 +180,7 @@ protected Map<String, String> resolveSecuritySchemes()
 
     protected <C> BindingConfigBuilder<C> injectMetrics(
         BindingConfigBuilder<C> binding,
-        List<MetricRefConfig> metricRefs,
-        String protocol)
+        List<MetricRefConfig> metricRefs)
     {
         if (metricRefs != null && !metricRefs.isEmpty())
         {