Use per worker registration for composite namespaces (#910)

aklivity · Apr 11, 2024 · 6a8f8a4 · 6a8f8a4
1 parent 10f8961
commit 6a8f8a4
Show file tree

Hide file tree

Showing 28 changed files with 278 additions and 357 deletions.
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
@@ -71,6 +71,7 @@ public final class AsyncapiBindingConfig
     private final Map<CharSequence, String> operationIds;
     private final LongFunction<CatalogHandler> supplyCatalog;
     private final ToLongFunction<String> resolveId;
+    private final Consumer<NamespaceConfig> attach;
     private final Consumer<NamespaceConfig> detach;
     private final long overrideRouteId;
     private final HttpHeaderHelper helper;
@@ -80,6 +81,8 @@ public AsyncapiBindingConfig(
         BindingConfig binding,
         AsyncapiNamespaceGenerator namespaceGenerator,
         LongFunction<CatalogHandler> supplyCatalog,
+        Consumer<NamespaceConfig> attachComposite,
+        Consumer<NamespaceConfig> detachComposite,
         long overrideRouteId)
     {
         this.id = binding.id;
@@ -99,7 +102,8 @@ public AsyncapiBindingConfig(
         this.operationIds = new TreeMap<>(CharSequence::compare);
         this.helper = new HttpHeaderHelper();
         this.parser = new AsyncapiParser();
-        this.detach = binding.detach;
+        this.attach = attachComposite;
+        this.detach = detachComposite;
         this.routes = binding.routes.stream().map(r -> new AsyncapiRouteConfig(r, schemaIdsByApiId::get)).collect(toList());
     }
 
@@ -218,9 +222,9 @@ private void attachProxyBinding(
                         (existingValue, newValue) -> existingValue,
                         Object2ObjectHashMap::new));
 
-        final NamespaceConfig composite = binding.attach.apply(namespaceGenerator.generateProxy(binding, asyncapis,
-            schemaIdsByApiId::get));
-
+        final NamespaceConfig composite = namespaceGenerator.generateProxy(binding, asyncapis, schemaIdsByApiId::get);
+        composite.readURL = binding.readURL;
+        attach.accept(composite);
         for (AsyncapiSchemaConfig config : configs)
         {
             Asyncapi asyncapi = config.asyncapi;
@@ -235,7 +239,9 @@ private void attachServerClientBinding(
         for (AsyncapiSchemaConfig config : configs)
         {
             Asyncapi asyncapi = config.asyncapi;
-            final NamespaceConfig composite = binding.attach.apply(namespaceGenerator.generate(binding, asyncapi));
+            final NamespaceConfig composite = namespaceGenerator.generate(binding, asyncapi);
+            composite.readURL = binding.readURL;
+            attach.accept(composite);
             updateNamespace(config, composite, asyncapi);
         }
     }

diff --git a/...ava/io/aklivity/zilla/runtime/binding/asyncapi/internal/stream/AsyncapiClientFactory.java b/...ava/io/aklivity/zilla/runtime/binding/asyncapi/internal/stream/AsyncapiClientFactory.java
@@ -14,6 +14,7 @@
  */
 package io.aklivity.zilla.runtime.binding.asyncapi.internal.stream;
 
+import java.util.function.Consumer;
 import java.util.function.LongFunction;
 import java.util.function.LongSupplier;
 import java.util.function.LongUnaryOperator;
@@ -43,6 +44,7 @@
 import io.aklivity.zilla.runtime.engine.buffer.BufferPool;
 import io.aklivity.zilla.runtime.engine.catalog.CatalogHandler;
 import io.aklivity.zilla.runtime.engine.config.BindingConfig;
+import io.aklivity.zilla.runtime.engine.config.NamespaceConfig;
 
 public final class AsyncapiClientFactory implements AsyncapiStreamFactory
 {
@@ -77,6 +79,8 @@ public final class AsyncapiClientFactory implements AsyncapiStreamFactory
     private final LongUnaryOperator supplyReplyId;
     private final LongSupplier supplyTraceId;
     private final LongFunction<CatalogHandler> supplyCatalog;
+    private final Consumer<NamespaceConfig> attachComposite;
+    private final Consumer<NamespaceConfig> detachComposite;
     private final Long2ObjectHashMap<AsyncapiBindingConfig> bindings;
     private final int asyncapiTypeId;
     private final int mqttTypeId;
@@ -98,6 +102,8 @@ public AsyncapiClientFactory(
         this.supplyReplyId = context::supplyReplyId;
         this.supplyTraceId = context::supplyTraceId;
         this.supplyCatalog = context::supplyCatalog;
+        this.attachComposite = context::attachComposite;
+        this.detachComposite = context::detachComposite;
         this.bindings = new Long2ObjectHashMap<>();
         this.asyncapiTypeId = context.supplyTypeId(AsyncapiBinding.NAME);
         this.mqttTypeId = context.supplyTypeId(MQTT_TYPE_NAME);
@@ -120,7 +126,7 @@ public void attach(
         BindingConfig binding)
     {
         AsyncapiBindingConfig asyncapiBinding = new AsyncapiBindingConfig(binding, namespaceGenerator, supplyCatalog,
-            config.targetRouteId());
+            attachComposite, detachComposite, config.targetRouteId());
         bindings.put(binding.id, asyncapiBinding);
 
         asyncapiBinding.attach(binding);

diff --git a/...java/io/aklivity/zilla/runtime/binding/asyncapi/internal/stream/AsyncapiProxyFactory.java b/...java/io/aklivity/zilla/runtime/binding/asyncapi/internal/stream/AsyncapiProxyFactory.java
@@ -15,6 +15,7 @@
 package io.aklivity.zilla.runtime.binding.asyncapi.internal.stream;
 
 import java.util.Optional;
+import java.util.function.Consumer;
 import java.util.function.Function;
 import java.util.function.LongFunction;
 import java.util.function.LongSupplier;
@@ -47,6 +48,7 @@
 import io.aklivity.zilla.runtime.engine.buffer.BufferPool;
 import io.aklivity.zilla.runtime.engine.catalog.CatalogHandler;
 import io.aklivity.zilla.runtime.engine.config.BindingConfig;
+import io.aklivity.zilla.runtime.engine.config.NamespaceConfig;
 
 public final class AsyncapiProxyFactory implements AsyncapiStreamFactory
 {
@@ -80,6 +82,8 @@ public final class AsyncapiProxyFactory implements AsyncapiStreamFactory
     private final LongSupplier supplyTraceId;
     private final Function<String, Integer> supplyTypeId;
     private final LongFunction<CatalogHandler> supplyCatalog;
+    private final Consumer<NamespaceConfig> attachComposite;
+    private final Consumer<NamespaceConfig> detachComposite;
     private final Long2ObjectHashMap<AsyncapiBindingConfig> bindings;
     private final Long2LongHashMap apiIds;
     private final AsyncapiConfiguration config;
@@ -102,6 +106,8 @@ public AsyncapiProxyFactory(
         this.supplyTypeId = context::supplyTypeId;
         this.supplyTraceId = context::supplyTraceId;
         this.supplyCatalog = context::supplyCatalog;
+        this.attachComposite = context::attachComposite;
+        this.detachComposite = context::detachComposite;
         this.bindings = new Long2ObjectHashMap<>();
         this.apiIds = new Long2LongHashMap(-1);
         this.asyncapiTypeId = context.supplyTypeId(AsyncapiBinding.NAME);
@@ -119,7 +125,7 @@ public void attach(
         BindingConfig binding)
     {
         AsyncapiBindingConfig asyncapiBinding = new AsyncapiBindingConfig(binding, namespaceGenerator,
-            supplyCatalog, config.targetRouteId());
+            supplyCatalog, attachComposite, detachComposite, config.targetRouteId());
         bindings.put(binding.id, asyncapiBinding);
 
         asyncapiBinding.attach(binding);

diff --git a/...ava/io/aklivity/zilla/runtime/binding/asyncapi/internal/stream/AsyncapiServerFactory.java b/...ava/io/aklivity/zilla/runtime/binding/asyncapi/internal/stream/AsyncapiServerFactory.java
@@ -14,6 +14,7 @@
  */
 package io.aklivity.zilla.runtime.binding.asyncapi.internal.stream;
 
+import java.util.function.Consumer;
 import java.util.function.Function;
 import java.util.function.LongFunction;
 import java.util.function.LongSupplier;
@@ -47,6 +48,7 @@
 import io.aklivity.zilla.runtime.engine.buffer.BufferPool;
 import io.aklivity.zilla.runtime.engine.catalog.CatalogHandler;
 import io.aklivity.zilla.runtime.engine.config.BindingConfig;
+import io.aklivity.zilla.runtime.engine.config.NamespaceConfig;
 
 public final class AsyncapiServerFactory implements AsyncapiStreamFactory
 {
@@ -86,6 +88,8 @@ public final class AsyncapiServerFactory implements AsyncapiStreamFactory
     private final LongSupplier supplyTraceId;
     private final Function<String, Integer> supplyTypeId;
     private final LongFunction<CatalogHandler> supplyCatalog;
+    private final Consumer<NamespaceConfig> attachComposite;
+    private final Consumer<NamespaceConfig> detachComposite;
     private final Long2ObjectHashMap<AsyncapiBindingConfig> bindings;
     private final int asyncapiTypeId;
     private final int mqttTypeId;
@@ -108,6 +112,8 @@ public AsyncapiServerFactory(
         this.supplyTypeId = context::supplyTypeId;
         this.supplyTraceId = context::supplyTraceId;
         this.supplyCatalog = context::supplyCatalog;
+        this.attachComposite = context::attachComposite;
+        this.detachComposite = context::detachComposite;
         this.bindings = new Long2ObjectHashMap<>();
         this.asyncapiTypeId = context.supplyTypeId(AsyncapiBinding.NAME);
         this.mqttTypeId = context.supplyTypeId(MQTT_TYPE_NAME);
@@ -131,7 +137,7 @@ public void attach(
         BindingConfig binding)
     {
         AsyncapiBindingConfig asyncapiBinding = new AsyncapiBindingConfig(binding, namespaceGenerator, supplyCatalog,
-            config.targetRouteId());
+            attachComposite, detachComposite, config.targetRouteId());
         bindings.put(binding.id, asyncapiBinding);
 
         asyncapiBinding.attach(binding);

diff --git a/.../zilla/runtime/binding/openapi/asyncapi/internal/config/OpenapiAsyncapiBindingConfig.java b/.../zilla/runtime/binding/openapi/asyncapi/internal/config/OpenapiAsyncapiBindingConfig.java
@@ -62,6 +62,7 @@ public final class OpenapiAsyncapiBindingConfig
     private final Object2LongHashMap<String> asyncapiSchemaIdsByApiId;
     private final AsyncapiParser asyncapiParser;
     private final OpenapiParser openapiParser;
+    private final Consumer<NamespaceConfig> attach;
     private final Consumer<NamespaceConfig> detach;
 
     private NamespaceConfig composite;
@@ -70,14 +71,17 @@ public final class OpenapiAsyncapiBindingConfig
     public OpenapiAsyncapiBindingConfig(
         BindingConfig binding,
         OpenapiAsyncNamespaceGenerator namespaceGenerator,
-        LongFunction<CatalogHandler> supplyCatalog)
+        LongFunction<CatalogHandler> supplyCatalog,
+        Consumer<NamespaceConfig> attachComposite,
+        Consumer<NamespaceConfig> detachComposite)
     {
         this.id = binding.id;
         this.name = binding.name;
         this.kind = binding.kind;
         this.options = (OpenapiAsyncapiOptionsConfig) binding.options;
         this.resolvedIds = binding.resolveId;
-        this.detach = binding.detach;
+        this.attach = attachComposite;
+        this.detach = detachComposite;
         this.openapiSchemaIdsByApiId = new Object2LongHashMap<>(-1);
         this.asyncapiSchemaIdsByApiId = new Object2LongHashMap<>(-1);
         this.compositeResolvedIds = new Long2LongHashMap(-1);
@@ -139,8 +143,9 @@ public void attach(
                     (e, n) -> e,
                     Object2ObjectHashMap::new));
 
-        this.composite = binding.attach.apply(namespaceGenerator.generate(binding, openapis,
-            asyncapis, openapiSchemaIdsByApiId::get));
+        this.composite = namespaceGenerator.generate(binding, openapis, asyncapis, openapiSchemaIdsByApiId::get);
+        this.composite.readURL = binding.readURL;
+        attach.accept(this.composite);
 
         BindingConfig mappingBinding = composite.bindings.stream()
             .filter(b -> b.type.equals("http-kafka")).findFirst().get();

diff --git a/.../zilla/runtime/binding/openapi/asyncapi/internal/streams/OpenapiAsyncapiProxyFactory.java b/.../zilla/runtime/binding/openapi/asyncapi/internal/streams/OpenapiAsyncapiProxyFactory.java
@@ -14,6 +14,7 @@
  */
 package io.aklivity.zilla.runtime.binding.openapi.asyncapi.internal.streams;
 
+import java.util.function.Consumer;
 import java.util.function.LongFunction;
 import java.util.function.LongSupplier;
 import java.util.function.LongUnaryOperator;
@@ -47,6 +48,7 @@
 import io.aklivity.zilla.runtime.engine.buffer.BufferPool;
 import io.aklivity.zilla.runtime.engine.catalog.CatalogHandler;
 import io.aklivity.zilla.runtime.engine.config.BindingConfig;
+import io.aklivity.zilla.runtime.engine.config.NamespaceConfig;
 
 public final class OpenapiAsyncapiProxyFactory implements OpenapiAsyncapiStreamFactory
 {
@@ -83,6 +85,8 @@ public final class OpenapiAsyncapiProxyFactory implements OpenapiAsyncapiStreamF
     private final LongUnaryOperator supplyReplyId;
     private final LongSupplier supplyTraceId;
     private final LongFunction<CatalogHandler> supplyCatalog;
+    private final Consumer<NamespaceConfig> attachComposite;
+    private final Consumer<NamespaceConfig> detachComposite;
     private final Long2ObjectHashMap<OpenapiAsyncapiBindingConfig> bindings;
     private final Long2LongHashMap apiIds;
     private final OpenapiAsyncapiConfiguration config;
@@ -104,6 +108,8 @@ public OpenapiAsyncapiProxyFactory(
         this.supplyReplyId = context::supplyReplyId;
         this.supplyTraceId = context::supplyTraceId;
         this.supplyCatalog = context::supplyCatalog;
+        this.attachComposite = context::attachComposite;
+        this.detachComposite = context::detachComposite;
         this.bindings = new Long2ObjectHashMap<>();
         this.apiIds = new Long2LongHashMap(-1);
         this.openapiTypeId = context.supplyTypeId(OpenapiBinding.NAME);
@@ -122,7 +128,7 @@ public void attach(
         BindingConfig binding)
     {
         OpenapiAsyncapiBindingConfig apiBinding = new OpenapiAsyncapiBindingConfig(binding,
-            namespaceGenerator, supplyCatalog);
+            namespaceGenerator, supplyCatalog, attachComposite, detachComposite);
         bindings.put(binding.id, apiBinding);
 
         apiBinding.attach(binding);

diff --git a/.../java/io/aklivity/zilla/runtime/binding/openapi/internal/config/OpenapiBindingConfig.java b/.../java/io/aklivity/zilla/runtime/binding/openapi/internal/config/OpenapiBindingConfig.java
@@ -69,6 +69,7 @@ public final class OpenapiBindingConfig
     private final Long2LongHashMap apiIdsByNamespaceId;
     private final HttpHeaderHelper helper;
     private final OpenapiParser parser;
+    private final Consumer<NamespaceConfig> attach;
     private final Consumer<NamespaceConfig> detach;
     private final Long2LongHashMap resolvedIds;
     private final Object2ObjectHashMap<Matcher, OpenapiPathItem> paths;
@@ -79,6 +80,8 @@ public OpenapiBindingConfig(
         BindingConfig binding,
         OpenapiNamespaceGenerator namespaceGenerator,
         LongFunction<CatalogHandler> supplyCatalog,
+        Consumer<NamespaceConfig> attachComposite,
+        Consumer<NamespaceConfig> detachComposite,
         long overrideRouteId)
     {
         this.id = binding.id;
@@ -96,7 +99,8 @@ public OpenapiBindingConfig(
         this.httpOrigins = new IntHashSet(-1);
         this.parser = new OpenapiParser();
         this.helper = new HttpHeaderHelper();
-        this.detach = binding.detach;
+        this.attach = attachComposite;
+        this.detach = detachComposite;
         this.routes = binding.routes.stream().map(OpenapiRouteConfig::new).collect(toList());
 
         Map<CharSequence, Function<OpenapiPathItem, String>> resolversByMethod = new TreeMap<>(CharSequence::compare);
@@ -118,7 +122,9 @@ public void attach(
         configs.forEach(c ->
         {
             Openapi openapi = c.openapi;
-            final NamespaceConfig composite = binding.attach.apply(namespaceGenerator.generate(binding, openapi));
+            final NamespaceConfig composite = namespaceGenerator.generate(binding, openapi);
+            composite.readURL = binding.readURL;
+            attach.accept(composite);
             composites.put(c.schemaId, composite);
             openapi.paths.forEach((k, v) ->
             {

diff --git a/...java/io/aklivity/zilla/runtime/binding/openapi/internal/streams/OpenapiClientFactory.java b/...java/io/aklivity/zilla/runtime/binding/openapi/internal/streams/OpenapiClientFactory.java
@@ -14,6 +14,7 @@
  */
 package io.aklivity.zilla.runtime.binding.openapi.internal.streams;
 
+import java.util.function.Consumer;
 import java.util.function.LongFunction;
 import java.util.function.LongSupplier;
 import java.util.function.LongUnaryOperator;
@@ -43,6 +44,7 @@
 import io.aklivity.zilla.runtime.engine.buffer.BufferPool;
 import io.aklivity.zilla.runtime.engine.catalog.CatalogHandler;
 import io.aklivity.zilla.runtime.engine.config.BindingConfig;
+import io.aklivity.zilla.runtime.engine.config.NamespaceConfig;
 
 public final class OpenapiClientFactory implements OpenapiStreamFactory
 {
@@ -79,11 +81,12 @@ public final class OpenapiClientFactory implements OpenapiStreamFactory
     private final LongUnaryOperator supplyReplyId;
     private final LongSupplier supplyTraceId;
     private final LongFunction<CatalogHandler> supplyCatalog;
+    private final Consumer<NamespaceConfig> attachComposite;
+    private final Consumer<NamespaceConfig> detachComposite;
     private final Long2ObjectHashMap<OpenapiBindingConfig> bindings;
     private final int openapiTypeId;
     private final int httpTypeId;
 
-
     public OpenapiClientFactory(
         OpenapiConfiguration config,
         EngineContext context)
@@ -97,6 +100,8 @@ public OpenapiClientFactory(
         this.supplyReplyId = context::supplyReplyId;
         this.supplyTraceId = context::supplyTraceId;
         this.supplyCatalog = context::supplyCatalog;
+        this.attachComposite = context::attachComposite;
+        this.detachComposite = context::detachComposite;
         this.namespaceGenerator = new OpenapiClientNamespaceGenerator();
         this.bindings = new Long2ObjectHashMap<>();
         this.openapiTypeId = context.supplyTypeId(OpenapiBinding.NAME);
@@ -120,7 +125,7 @@ public void attach(
         BindingConfig binding)
     {
         OpenapiBindingConfig openapiBinding = new OpenapiBindingConfig(binding, namespaceGenerator, supplyCatalog,
-            config.targetRouteId());
+            attachComposite, detachComposite, config.targetRouteId());
         bindings.put(binding.id, openapiBinding);
 
         openapiBinding.attach(binding);

diff --git a/...java/io/aklivity/zilla/runtime/binding/openapi/internal/streams/OpenapiServerFactory.java b/...java/io/aklivity/zilla/runtime/binding/openapi/internal/streams/OpenapiServerFactory.java
@@ -14,6 +14,7 @@
  */
 package io.aklivity.zilla.runtime.binding.openapi.internal.streams;
 
+import java.util.function.Consumer;
 import java.util.function.LongFunction;
 import java.util.function.LongSupplier;
 import java.util.function.LongUnaryOperator;
@@ -45,6 +46,7 @@
 import io.aklivity.zilla.runtime.engine.buffer.BufferPool;
 import io.aklivity.zilla.runtime.engine.catalog.CatalogHandler;
 import io.aklivity.zilla.runtime.engine.config.BindingConfig;
+import io.aklivity.zilla.runtime.engine.config.NamespaceConfig;
 
 public final class OpenapiServerFactory implements OpenapiStreamFactory
 {
@@ -82,6 +84,8 @@ public final class OpenapiServerFactory implements OpenapiStreamFactory
     private final LongUnaryOperator supplyReplyId;
     private final LongSupplier supplyTraceId;
     private final LongFunction<CatalogHandler> supplyCatalog;
+    private final Consumer<NamespaceConfig> attachComposite;
+    private final Consumer<NamespaceConfig> detachComposite;
     private final Long2ObjectHashMap<OpenapiBindingConfig> bindings;
     private final int openapiTypeId;
     private final int httpTypeId;
@@ -99,6 +103,8 @@ public OpenapiServerFactory(
         this.supplyReplyId = context::supplyReplyId;
         this.supplyTraceId = context::supplyTraceId;
         this.supplyCatalog = context::supplyCatalog;
+        this.attachComposite = context::attachComposite;
+        this.detachComposite = context::detachComposite;
         this.namespaceGenerator = new OpenapiServerNamespaceGenerator();
         this.bindings = new Long2ObjectHashMap<>();
         this.openapiTypeId = context.supplyTypeId(OpenapiBinding.NAME);
@@ -122,7 +128,7 @@ public void attach(
         BindingConfig binding)
     {
         OpenapiBindingConfig openapiBinding = new OpenapiBindingConfig(binding, namespaceGenerator, supplyCatalog,
-            config.targetRouteId());
+            attachComposite, detachComposite, config.targetRouteId());
         bindings.put(binding.id, openapiBinding);
 
         openapiBinding.attach(binding);