WIP

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

@@ -20,7 +20,6 @@ catalogs:
   catalog0:
     type: test
     options:
-      id: 1
       subject: pet
       schema: |
         openapi: "3.0.0"
@@ -31,7 +30,6 @@ catalogs:
             name: MIT
         servers:
           - url: http://localhost:8080
-          - url: https://localhost:9090
         paths:
           /pets:
             get:
@@ -148,8 +146,7 @@ bindings:
   composite0:
     type: openapi
     kind: server
-    options:
-      catalog:
-        catalog0:
-          - subject: pet
+    catalog:
+      catalog0:
+        - subject: pet
     exit: openapi0

incubator/binding-openapi/src/main/java/io/aklivity/zilla/runtime/binding/openapi/internal/config/OpenapiBindingConfig.java

@@ -15,9 +15,6 @@
 package io.aklivity.zilla.runtime.binding.openapi.internal.config;
 
 import static java.util.Collections.unmodifiableMap;
-import static java.util.stream.Collector.of;
-import static java.util.stream.Collector.Characteristics.IDENTITY_FINISH;
-import static java.util.stream.Collectors.toCollection;
 import static java.util.stream.Collectors.toList;
 
 import java.util.ArrayList;
@@ -33,6 +30,7 @@
 
 import org.agrona.AsciiSequenceView;
 import org.agrona.DirectBuffer;
+import org.agrona.collections.Int2ObjectHashMap;
 import org.agrona.collections.IntHashSet;
 import org.agrona.collections.Long2LongHashMap;
 import org.agrona.collections.Object2ObjectHashMap;
@@ -66,11 +64,12 @@ public final class OpenapiBindingConfig
     private final LongFunction<CatalogHandler> supplyCatalog;
     private final long overrideRouteId;
     private final IntHashSet httpOrigins;
+    private final Long2LongHashMap apiIdsByNamespaceId;
     private final OpenapiParser parser;
     private final Consumer<NamespaceConfig> detach;
     private final Long2LongHashMap resolvedIds;
     private final Object2ObjectHashMap<Matcher, OpenapiPathItem> paths;
-    private final List<NamespaceConfig> composites;
+    private final Int2ObjectHashMap<NamespaceConfig> composites;
     private final Map<CharSequence, Function<OpenapiPathItem, String>> resolversByMethod;
 
     public OpenapiBindingConfig(
@@ -86,15 +85,19 @@ public OpenapiBindingConfig(
         this.overrideRouteId = overrideRouteId;
         this.options = OpenapiOptionsConfig.class.cast(binding.options);
         this.paths = new Object2ObjectHashMap<>();
-        this.composites = new ArrayList<>();
+        this.composites = new Int2ObjectHashMap<>();
+        this.resolvedIds = new Long2LongHashMap(-1);
+        this.apiIdsByNamespaceId = new Long2LongHashMap(-1);
+        this.httpOrigins = new IntHashSet(-1);
         this.parser = new OpenapiParser();
         this.detach = binding.detach;
+
         List<OpenapiConfig> configs = convertToOpenapi(binding.catalogs);
         configs.forEach(c ->
         {
             Openapi openapi = c.openapi;
             final NamespaceConfig composite = binding.attach.apply(compositeBinding.composite(binding, openapi));
-            composites.add(composite);
+            composites.put(c.schemaId, composite);
             openapi.paths.forEach((k, v) ->
             {
                 String regex = k.replaceAll("\\{[^/]+}", "[^/]+");
@@ -106,23 +109,22 @@ public OpenapiBindingConfig(
 
         this.routes = binding.routes.stream().map(OpenapiRouteConfig::new).collect(toList());
 
-        this.resolvedIds = binding.composites.values().stream()
-            .map(c -> c.bindings)
-            .flatMap(List::stream)
-            .filter(b -> b.type.equals("http"))
-            .collect(of(
-                () -> new Long2LongHashMap(-1),
-                (m, r) -> m.put(0, r.id), //options.openapis.stream().findFirst().get().apiId
-                (m, r) -> m,
-                IDENTITY_FINISH
-            ));
-
-        this.httpOrigins = binding.composites.values().stream()
-            .map(c -> c.bindings)
-            .flatMap(List::stream)
-            .filter(b -> b.type.equals("http"))
-            .map(b -> NamespacedId.namespaceId(b.id))
-            .collect(toCollection(IntHashSet::new));
+        composites.forEach((k, v) ->
+        {
+            List<BindingConfig> http = v.bindings.stream().filter(b -> b.type.equals("http")).collect(toList());
+            http.stream()
+                .map(b -> b.routes)
+                .flatMap(List::stream)
+                .forEach(r -> resolvedIds.put(k, r.id));
+            http.stream()
+                .map(b -> NamespacedId.namespaceId(b.id))
+                .forEach(n ->
+                {
+                    httpOrigins.add(n);
+                    apiIdsByNamespaceId.put(n, k);
+                });
+        });
+
         this.helper = new HttpHeaderHelper();
 
         Map<CharSequence, Function<OpenapiPathItem, String>> resolversByMethod = new TreeMap<>(CharSequence::compare);
@@ -139,7 +141,7 @@ public OpenapiBindingConfig(
 
     public void detach()
     {
-        composites.forEach(detach);
+        composites.forEach((k, v) -> detach.accept(v));
         composites.clear();
     }
 
@@ -149,6 +151,12 @@ public boolean isCompositeNamespace(
         return httpOrigins.contains(namespaceId);
     }
 
+    public long resolveApiId(
+        long namespaceId)
+    {
+        return apiIdsByNamespaceId.get(namespaceId);
+    }
+
     public long resolveResolvedId(
         long apiId)
     {

incubator/binding-openapi/src/main/java/io/aklivity/zilla/runtime/binding/openapi/internal/config/OpenapiCompositeBinding.java

@@ -23,6 +23,7 @@
 import java.util.regex.Pattern;
 
 
+import io.aklivity.zilla.runtime.binding.openapi.config.OpenapiOptionsConfig;
 import io.aklivity.zilla.runtime.binding.openapi.internal.model.Openapi;
 import io.aklivity.zilla.runtime.engine.config.BindingConfig;
 import io.aklivity.zilla.runtime.engine.config.BindingConfigBuilder;
@@ -40,13 +41,16 @@ public abstract class OpenapiCompositeBinding
     protected static final String INLINE_CATALOG_TYPE = "inline";
     protected static final String VERSION_LATEST = "latest";
     protected static final Pattern JSON_CONTENT_TYPE = Pattern.compile("^application/(?:.+\\+)?json$");
+    protected static final OpenapiOptionsConfig EMPTY_OPTION = new OpenapiOptionsConfig(null, null, null);
 
     protected final Matcher jsonContentType = JSON_CONTENT_TYPE.matcher("");
     protected final Map<String, ModelConfig> models = Map.of(
         "string", StringModelConfig.builder().build(),
         "integer", Int32ModelConfig.builder().build()
     );
 
+
+
     public abstract NamespaceConfig composite(
         BindingConfig binding,
         Openapi openapi);

incubator/binding-openapi/src/main/java/io/aklivity/zilla/runtime/binding/openapi/internal/config/OpenapiServerCompositeBinding.java

@@ -67,7 +67,7 @@ public NamespaceConfig composite(
         BindingConfig binding,
         Openapi openapi)
     {
-        final OpenapiOptionsConfig options = (OpenapiOptionsConfig) binding.options;
+        final OpenapiOptionsConfig options = binding.options != null ? (OpenapiOptionsConfig) binding.options : EMPTY_OPTION;
         final List<MetricRefConfig> metricRefs = binding.telemetryRef != null ?
             binding.telemetryRef.metricRefs : emptyList();
 

incubator/binding-openapi/src/main/java/io/aklivity/zilla/runtime/binding/openapi/internal/streams/OpenapiServerFactory.java

@@ -156,28 +156,32 @@ public MessageConsumer newStream(
 
         MessageConsumer newStream = null;
 
-        if (binding != null && binding.isCompositeNamespace(NamespacedId.namespaceId(originId)))
+        if (binding != null)
         {
+            final int namespaceId = NamespacedId.namespaceId(originId);
+            if (binding.isCompositeNamespace(namespaceId))
+            {
 
-            final OpenapiRouteConfig route = binding.resolve(authorization);
+                final OpenapiRouteConfig route = binding.resolve(authorization);
+
+                if (route != null)
+                {
+                    final String operationId = binding.resolveOperationId(httpBeginEx);
+                    final long apiId = binding.resolveApiId(namespaceId);
+
+                    newStream = new HttpStream(
+                        receiver,
+                        originId,
+                        routedId,
+                        initialId,
+                        affinity,
+                        authorization,
+                        route.id,
+                        apiId,
+                        operationId)::onHttpMessage;
+                }
 
-            if (route != null)
-            {
-                final long apiId = 0L; //binding.options.openapis.get(0).apiId;
-                final String operationId = binding.resolveOperationId(httpBeginEx);
-
-                newStream = new HttpStream(
-                    receiver,
-                    originId,
-                    routedId,
-                    initialId,
-                    affinity,
-                    authorization,
-                    route.id,
-                    apiId,
-                    operationId)::onHttpMessage;
             }
-
         }
 
         return newStream;

incubator/binding-openapi/src/test/java/io/aklivity/zilla/runtime/binding/openapi/internal/config/OpenapiOptionsConfigAdapterTest.java

@@ -37,10 +37,7 @@
 import org.mockito.junit.MockitoJUnit;
 import org.mockito.junit.MockitoRule;
 
-import io.aklivity.zilla.runtime.binding.openapi.config.OpenapiConfig;
 import io.aklivity.zilla.runtime.binding.openapi.config.OpenapiOptionsConfig;
-import io.aklivity.zilla.runtime.binding.openapi.internal.model.Openapi;
-import io.aklivity.zilla.runtime.binding.openapi.internal.model.OpenapiPathItem;
 import io.aklivity.zilla.runtime.binding.tcp.config.TcpOptionsConfig;
 import io.aklivity.zilla.runtime.binding.tls.config.TlsOptionsConfig;
 import io.aklivity.zilla.runtime.engine.config.ConfigAdapterContext;
@@ -101,20 +98,12 @@ public void shouldReadOptions()
             "          }" +
             "        }" +
             "      }" +
-            "    }," +
-            "    \"specs\": {" +
-            "      \"openapi-id\": \"openapi/petstore.yaml\"" +
             "    }" +
             "  }";
 
         OpenapiOptionsConfig options = jsonb.fromJson(text, OpenapiOptionsConfig.class);
-        OpenapiConfig openapi = options.openapis.stream().findFirst().get();
-        OpenapiPathItem path = openapi.openapi.paths.get("/pets");
 
         assertThat(options, not(nullValue()));
-        assertThat(path.post, not(nullValue()));
-        assertThat(options.tls, not(nullValue()));
-        assertThat(options.http, not(nullValue()));
     }
 
     @Test
@@ -134,8 +123,7 @@ public void shouldWriteOptions()
             .sni(asList("example.net"))
             .build();
 
-        OpenapiOptionsConfig options = new OpenapiOptionsConfig(tcp, tls, null, asList(
-            new OpenapiConfig("openapi-id", 1L, "openapi/petstore.yaml", new Openapi())));
+        OpenapiOptionsConfig options = new OpenapiOptionsConfig(tcp, tls, null);
 
         String text = jsonb.toJson(options);
 

runtime/engine/src/main/java/io/aklivity/zilla/runtime/engine/internal/registry/EngineManager.java

@@ -138,8 +138,8 @@ public EngineConfig reconfigure(
 
                 try
                 {
-                    register(newConfig);
                     current = newConfig;
+                    register(newConfig);
                 }
                 catch (Exception ex)
                 {
@@ -403,8 +403,8 @@ private void register(
     {
         dispatchers.stream()
             .map(d -> d.attach(namespace))
-            .reduce(CompletableFuture::allOf)
-            .ifPresent(CompletableFuture::join);
+            .reduce(CompletableFuture::allOf);
+            //.ifPresent(CompletableFuture::join);
     }
 
     private void unregister(
@@ -414,8 +414,8 @@ private void unregister(
         {
             dispatchers.stream()
                 .map(d -> d.detach(namespace))
-                .reduce(CompletableFuture::allOf)
-                .ifPresent(CompletableFuture::join);
+                .reduce(CompletableFuture::allOf);
+                //.ifPresent(CompletableFuture::join);
         }
     }