feat(engine): add RouterFactorySpi for pluggable stream factory composition (#1757)

Author: jfallows

runtime/engine/AGENTS.md

@@ -46,7 +46,7 @@ default implementation of the new method or the build will fail.
 ## Test implementations for engine concepts
 
 Every engine concept (binding, guard, vault, catalog, exporter, metric group,
-model, resolver) has a minimal **test implementation** that lives in this
+model, resolver, router) has a minimal **test implementation** that lives in this
 module's test sources under
 `src/test/java/.../engine/test/internal/<concept>/`. For example:
 
@@ -60,6 +60,7 @@ module's test sources under
 | metric group | `TestMetricGroupFactorySpi`, `TestMetricGroup` |
 | model | `TestModelFactorySpi`, `TestModel`, `TestModelContext` |
 | resolver | `TestResolverFactorySpi`, `TestResolverSpi` |
+| router | `TestRouterFactorySpi`, `TestRouter`, `TestRouterContext` |
 
 The engine module is built with Maven's `test-jar` packaging so these classes
 are published as `engine:<version>:test-jar`. Every `specs/*.spec` module

runtime/engine/pom.xml

@@ -288,6 +288,7 @@
                 <include>io/aklivity/zilla/runtime/engine/test/internal/metrics/**/*.class</include>
                 <include>io/aklivity/zilla/runtime/engine/test/internal/model/**/*.class</include>
                 <include>io/aklivity/zilla/runtime/engine/test/internal/resolver/**/*.class</include>
+                <include>io/aklivity/zilla/runtime/engine/test/internal/router/**/*.class</include>
                 <include>io/aklivity/zilla/runtime/engine/test/internal/store/**/*.class</include>
                 <include>io/aklivity/zilla/runtime/engine/test/internal/vault/**/*.class</include>
                 <include>io/aklivity/zilla/runtime/engine/internal/concurrent/bench/**/*.class</include>

runtime/engine/src/main/java/io/aklivity/zilla/runtime/engine/Engine.java

@@ -58,6 +58,7 @@
 import io.aklivity.zilla.runtime.engine.catalog.Catalog;
 import io.aklivity.zilla.runtime.engine.config.KindConfig;
 import io.aklivity.zilla.runtime.engine.config.NamespaceConfig;
+import io.aklivity.zilla.runtime.engine.config.RouterConfig;
 import io.aklivity.zilla.runtime.engine.diagnostic.EngineDiagnosticsTask;
 import io.aklivity.zilla.runtime.engine.event.EventFormatterFactory;
 import io.aklivity.zilla.runtime.engine.exporter.Exporter;
@@ -77,6 +78,8 @@
 import io.aklivity.zilla.runtime.engine.metrics.MetricGroup;
 import io.aklivity.zilla.runtime.engine.model.Model;
 import io.aklivity.zilla.runtime.engine.namespace.NamespacedId;
+import io.aklivity.zilla.runtime.engine.router.Router;
+import io.aklivity.zilla.runtime.engine.router.RouterFactory;
 import io.aklivity.zilla.runtime.engine.store.Store;
 import io.aklivity.zilla.runtime.engine.vault.Vault;
 
@@ -99,6 +102,7 @@ public final class Engine implements Collector, AutoCloseable
     private final EngineConfiguration config;
     private final EngineManager manager;
     private final EngineDiagnosticsTask diagnostics;
+    private final RouterConfig routerConfig;
 
     private final EventWriter eventWriter;
     private final AtomicBoolean closed;
@@ -190,13 +194,20 @@ public final class Engine implements Collector, AutoCloseable
             }
         }
 
+        final Router router = RouterFactory.instantiate().create(config.router(), config);
+        final RouterConfig routerConfig = RouterConfig.builder()
+            .id(0L)
+            .name(router.name())
+            .build();
+        this.routerConfig = routerConfig;
+
         List<EngineWorker> workers = new ArrayList<>(workerCount);
         for (int workerIndex = 0; workerIndex < workerCount; workerIndex++)
         {
             EngineWorker worker =
                 new EngineWorker(config, tasks, labels, diagnoseOnError, tuning::affinity, bindings, exporters,
-                    guards, vaults, catalogs, models, metricGroups, stores, this, this::supplyEventReader,
-                    eventFormatterFactory, workerIndex, readonly, this::process, boss);
+                    guards, vaults, catalogs, models, metricGroups, stores, router, routerConfig, this,
+                    this::supplyEventReader, eventFormatterFactory, workerIndex, readonly, this::process, boss);
             workers.add(worker);
         }
         this.workers = workers;
@@ -391,6 +402,11 @@ public Clock clock()
         return config.clock();
     }
 
+    public RouterConfig routerConfig()
+    {
+        return routerConfig;
+    }
+
     public static EngineBuilder builder()
     {
         return new EngineBuilder();

runtime/engine/src/main/java/io/aklivity/zilla/runtime/engine/EngineConfiguration.java

@@ -107,6 +107,7 @@ public class EngineConfiguration extends Configuration
     public static final PropertyDef<ErrorReporter> ENGINE_ERROR_REPORTER;
     public static final PropertyDef<RevocationStrategy> ENGINE_CERTIFICATE_REVOCATION_STRATEGY;
     public static final PropertyDef<Path> ENGINE_DIAGNOSTICS_DIRECTORY;
+    public static final PropertyDef<String> ENGINE_ROUTER;
 
     private static final ConfigurationDef ENGINE_CONFIG;
 
@@ -175,6 +176,7 @@ public class EngineConfiguration extends Configuration
             EngineConfiguration::decodeRevocationStrategy, RevocationStrategy.NONE);
         ENGINE_DIAGNOSTICS_DIRECTORY = config.property(Path.class, "diagnostics.directory",
             EngineConfiguration::decodeDiagnosticsDirectory, (String) null);
+        ENGINE_ROUTER = config.property("router", "engine");
         ENGINE_CONFIG = config;
     }
 
@@ -425,6 +427,11 @@ public Function<String, InetAddress[]> hostResolver()
         return ENGINE_HOST_RESOLVER.get(this)::resolve;
     }
 
+    public String router()
+    {
+        return ENGINE_ROUTER.get(this);
+    }
+
     private static int defaultTaskParallelism(
         Configuration config)
     {

runtime/engine/src/main/java/io/aklivity/zilla/runtime/engine/config/RouterConfig.java

@@ -0,0 +1,41 @@
+/*
+ * Copyright 2021-2024 Aklivity Inc.
+ *
+ * Aklivity licenses this file to you under the Apache License,
+ * version 2.0 (the "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at:
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+ * License for the specific language governing permissions and limitations
+ * under the License.
+ */
+package io.aklivity.zilla.runtime.engine.config;
+
+import static java.util.Objects.requireNonNull;
+import static java.util.function.Function.identity;
+
+public class RouterConfig
+{
+    public final long id;
+    public final String name;
+    public final OptionsConfig options;
+
+    public static RouterConfigBuilder<RouterConfig> builder()
+    {
+        return new RouterConfigBuilder<>(identity());
+    }
+
+    protected RouterConfig(
+        long id,
+        String name,
+        OptionsConfig options)
+    {
+        this.id = id;
+        this.name = requireNonNull(name);
+        this.options = options;
+    }
+}

runtime/engine/src/main/java/io/aklivity/zilla/runtime/engine/config/RouterConfigBuilder.java

@@ -0,0 +1,73 @@
+/*
+ * Copyright 2021-2024 Aklivity Inc.
+ *
+ * Aklivity licenses this file to you under the Apache License,
+ * version 2.0 (the "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at:
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+ * License for the specific language governing permissions and limitations
+ * under the License.
+ */
+package io.aklivity.zilla.runtime.engine.config;
+
+import java.util.function.Function;
+
+public final class RouterConfigBuilder<T> extends ConfigBuilder<T, RouterConfigBuilder<T>>
+{
+    private final Function<RouterConfig, T> mapper;
+
+    private long id;
+    private String name;
+    private OptionsConfig options;
+
+    RouterConfigBuilder(
+        Function<RouterConfig, T> mapper)
+    {
+        this.mapper = mapper;
+    }
+
+    @Override
+    @SuppressWarnings("unchecked")
+    protected Class<RouterConfigBuilder<T>> thisType()
+    {
+        return (Class<RouterConfigBuilder<T>>) getClass();
+    }
+
+    public RouterConfigBuilder<T> id(
+        long id)
+    {
+        this.id = id;
+        return this;
+    }
+
+    public RouterConfigBuilder<T> name(
+        String name)
+    {
+        this.name = name;
+        return this;
+    }
+
+    public <C extends ConfigBuilder<RouterConfigBuilder<T>, C>> C options(
+        Function<Function<OptionsConfig, RouterConfigBuilder<T>>, C> options)
+    {
+        return options.apply(this::options);
+    }
+
+    public RouterConfigBuilder<T> options(
+        OptionsConfig options)
+    {
+        this.options = options;
+        return this;
+    }
+
+    @Override
+    public T build()
+    {
+        return mapper.apply(new RouterConfig(id, name, options));
+    }
+}

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

@@ -0,0 +1,69 @@
+/*
+ * Copyright 2021-2024 Aklivity Inc.
+ *
+ * Aklivity licenses this file to you under the Apache License,
+ * version 2.0 (the "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at:
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+ * License for the specific language governing permissions and limitations
+ * under the License.
+ */
+package io.aklivity.zilla.runtime.engine.internal.registry;
+
+import java.util.function.Consumer;
+
+import io.aklivity.zilla.runtime.engine.Configuration;
+import io.aklivity.zilla.runtime.engine.binding.BindingHandler;
+import io.aklivity.zilla.runtime.engine.config.NamespaceConfig;
+import io.aklivity.zilla.runtime.engine.router.RouteableContext;
+
+final class EngineRouteable implements RouteableContext
+{
+    private final Configuration config;
+    private final BindingHandler streamFactory;
+    private final Consumer<NamespaceConfig> attachComposite;
+    private final Consumer<NamespaceConfig> detachComposite;
+
+    EngineRouteable(
+        Configuration config,
+        BindingHandler streamFactory,
+        Consumer<NamespaceConfig> attachComposite,
+        Consumer<NamespaceConfig> detachComposite)
+    {
+        this.config = config;
+        this.streamFactory = streamFactory;
+        this.attachComposite = attachComposite;
+        this.detachComposite = detachComposite;
+    }
+
+    @Override
+    public Configuration config()
+    {
+        return config;
+    }
+
+    @Override
+    public BindingHandler streamFactory()
+    {
+        return streamFactory;
+    }
+
+    @Override
+    public void attachComposite(
+        NamespaceConfig composite)
+    {
+        attachComposite.accept(composite);
+    }
+
+    @Override
+    public void detachComposite(
+        NamespaceConfig composite)
+    {
+        detachComposite.accept(composite);
+    }
+}

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

@@ -0,0 +1,38 @@
+/*
+ * Copyright 2021-2024 Aklivity Inc.
+ *
+ * Aklivity licenses this file to you under the Apache License,
+ * version 2.0 (the "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at:
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+ * License for the specific language governing permissions and limitations
+ * under the License.
+ */
+package io.aklivity.zilla.runtime.engine.internal.registry;
+
+import io.aklivity.zilla.runtime.engine.router.RouteableContext;
+import io.aklivity.zilla.runtime.engine.router.Router;
+import io.aklivity.zilla.runtime.engine.router.RouterContext;
+
+public final class EngineRouter implements Router
+{
+    public static final String NAME = "engine";
+
+    @Override
+    public String name()
+    {
+        return NAME;
+    }
+
+    @Override
+    public RouterContext supply(
+        RouteableContext context)
+    {
+        return new EngineRouterContext(context.streamFactory());
+    }
+}

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

@@ -0,0 +1,44 @@
+/*
+ * Copyright 2021-2024 Aklivity Inc.
+ *
+ * Aklivity licenses this file to you under the Apache License,
+ * version 2.0 (the "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at:
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+ * License for the specific language governing permissions and limitations
+ * under the License.
+ */
+package io.aklivity.zilla.runtime.engine.internal.registry;
+
+import io.aklivity.zilla.runtime.engine.binding.BindingHandler;
+import io.aklivity.zilla.runtime.engine.config.RouterConfig;
+import io.aklivity.zilla.runtime.engine.router.RouterContext;
+
+final class EngineRouterContext implements RouterContext
+{
+    private final BindingHandler streamFactory;
+
+    EngineRouterContext(
+        BindingHandler streamFactory)
+    {
+        this.streamFactory = streamFactory;
+    }
+
+    @Override
+    public BindingHandler attach(
+        RouterConfig config)
+    {
+        return streamFactory;
+    }
+
+    @Override
+    public void detach(
+        long routerId)
+    {
+    }
+}