WIP fix ResourceResolver

incubator/catalog-filesystem/src/main/java/io/aklivity/zilla/runtime/catalog/filesystem/internal/FilesystemCatalogHandler.java

@@ -17,7 +17,6 @@
 import java.io.InputStream;
 import java.net.URL;
 import java.util.HashMap;
-import java.util.LinkedList;
 import java.util.List;
 import java.util.Map;
 import java.util.function.Function;
@@ -36,7 +35,6 @@ public class FilesystemCatalogHandler implements CatalogHandler
     private final FilesystemEventContext event;
     private final long catalogId;
     private final Function<String, URL> resolvePath;
-    private final List<String> resources;
 
     public FilesystemCatalogHandler(
         FilesystemOptionsConfig config,
@@ -50,11 +48,6 @@ public FilesystemCatalogHandler(
         this.resolvePath = context::resolvePath;
         this.catalogId = catalogId;
         registerSchema(config.subjects);
-        this.resources = new LinkedList<>();
-        for (FilesystemSchemaConfig subject : config.subjects)
-        {
-            resources.add(subject.path);
-        }
     }
 
     @Override
@@ -95,12 +88,6 @@ private void registerSchema(
         }
     }
 
-    @Override
-    public List<String> resources()
-    {
-        return resources;
-    }
-
     private int generateCRC32C(
         String schema)
     {

incubator/catalog-filesystem/src/main/java/io/aklivity/zilla/runtime/catalog/filesystem/internal/config/FilesystemOptionsConfig.java

@@ -14,6 +14,7 @@
  */
 package io.aklivity.zilla.runtime.catalog.filesystem.internal.config;
 
+import java.util.LinkedList;
 import java.util.List;
 import java.util.function.Function;
 
@@ -37,6 +38,18 @@ public static <T> FilesystemOptionsConfigBuilder<T> builder(
     public FilesystemOptionsConfig(
         List<FilesystemSchemaConfig> subjects)
     {
+        super(List.of(), resolveResources(subjects));
         this.subjects = subjects;
     }
+
+    private static List<String> resolveResources(
+        List<FilesystemSchemaConfig> subjects)
+    {
+        List<String> resources = new LinkedList<>();
+        for (FilesystemSchemaConfig subject : subjects)
+        {
+            resources.add(subject.path);
+        }
+        return resources;
+    }
 }

runtime/binding-http/src/main/java/io/aklivity/zilla/runtime/binding/http/config/HttpOptionsConfig.java

@@ -27,6 +27,7 @@
 
 import io.aklivity.zilla.runtime.binding.http.internal.types.String16FW;
 import io.aklivity.zilla.runtime.binding.http.internal.types.String8FW;
+import io.aklivity.zilla.runtime.engine.config.ModelConfig;
 import io.aklivity.zilla.runtime.engine.config.OptionsConfig;
 
 public final class HttpOptionsConfig extends OptionsConfig
@@ -55,41 +56,46 @@ public static <T> HttpOptionsConfigBuilder<T> builder(
         HttpAuthorizationConfig authorization,
         List<HttpRequestConfig> requests)
     {
-        super(requests != null && !requests.isEmpty()
-            ? requests.stream()
-                .flatMap(request -> Stream.concat(
-                    Stream.of(request.content),
-                    Stream.concat(
-                        request.headers != null
-                            ? request.headers.stream().flatMap(header -> Stream.of(header != null ? header.model : null))
-                            : Stream.empty(),
-                        Stream.concat(
-                            request.pathParams != null
-                                ? request.pathParams.stream().flatMap(param -> Stream.of(param != null ? param.model : null))
-                                : Stream.empty(),
-                            Stream.concat(
-                                    request.queryParams != null
-                                    ? request.queryParams.stream().flatMap(param -> Stream.of(param != null ? param.model : null))
-                                    : Stream.empty(),
-                                    Stream.concat(request.responses != null
-                                        ? request.responses.stream().flatMap(param -> Stream.of(param != null
-                                            ? param.content
-                                            : null))
-                                        : Stream.empty(), request.responses != null
-                                        ? request.responses.stream()
-                                            .flatMap(response -> response.headers != null
-                                                ? response.headers.stream()
-                                                    .flatMap(param -> Stream.of(param != null ? param.model : null))
-                                                : Stream.empty())
-                                        : Stream.empty())
-                            )))).filter(Objects::nonNull))
-                .collect(Collectors.toList())
-            : emptyList());
-
+        super(resolveModels(requests), List.of());
         this.versions = versions;
         this.overrides = overrides;
         this.access = access;
         this.authorization = authorization;
         this.requests = requests;
     }
+
+    private static List<ModelConfig> resolveModels(
+        List<HttpRequestConfig> requests)
+    {
+        return requests != null && !requests.isEmpty()
+            ? requests.stream()
+            .flatMap(request -> Stream.concat(
+                Stream.of(request.content),
+                Stream.concat(
+                    request.headers != null
+                        ? request.headers.stream().flatMap(header -> Stream.of(header != null ? header.model : null))
+                        : Stream.empty(),
+                    Stream.concat(
+                        request.pathParams != null
+                            ? request.pathParams.stream().flatMap(param -> Stream.of(param != null ? param.model : null))
+                            : Stream.empty(),
+                        Stream.concat(
+                            request.queryParams != null
+                                ? request.queryParams.stream().flatMap(param -> Stream.of(param != null ? param.model : null))
+                                : Stream.empty(),
+                            Stream.concat(request.responses != null
+                                ? request.responses.stream().flatMap(param -> Stream.of(param != null
+                                ? param.content
+                                : null))
+                                : Stream.empty(), request.responses != null
+                                ? request.responses.stream()
+                                .flatMap(response -> response.headers != null
+                                    ? response.headers.stream()
+                                    .flatMap(param -> Stream.of(param != null ? param.model : null))
+                                    : Stream.empty())
+                                : Stream.empty())
+                        )))).filter(Objects::nonNull))
+            .collect(Collectors.toList())
+            : emptyList();
+    }
 }

runtime/binding-kafka/src/main/java/io/aklivity/zilla/runtime/binding/kafka/config/KafkaOptionsConfig.java

@@ -23,6 +23,7 @@
 import java.util.function.Function;
 import java.util.stream.Stream;
 
+import io.aklivity.zilla.runtime.engine.config.ModelConfig;
 import io.aklivity.zilla.runtime.engine.config.OptionsConfig;
 
 public final class KafkaOptionsConfig extends OptionsConfig
@@ -49,15 +50,21 @@ public static <T> KafkaOptionsConfigBuilder<T> builder(
         List<KafkaServerConfig> servers,
         KafkaSaslConfig sasl)
     {
-        super(topics != null && !topics.isEmpty()
-            ? topics.stream()
-                .flatMap(t -> Stream.of(t.key, t.value))
-                .filter(Objects::nonNull)
-                .collect(toList())
-            : emptyList());
+        super(resolveModels(topics), List.of());
         this.bootstrap = bootstrap;
         this.topics = topics;
         this.servers = servers;
         this.sasl = sasl;
     }
+
+    private static List<ModelConfig> resolveModels(
+        List<KafkaTopicConfig> topics)
+    {
+        return topics != null && !topics.isEmpty()
+            ? topics.stream()
+            .flatMap(t -> Stream.of(t.key, t.value))
+            .filter(Objects::nonNull)
+            .collect(toList())
+            : emptyList();
+    }
 }

runtime/binding-mqtt/src/main/java/io/aklivity/zilla/runtime/binding/mqtt/config/MqttOptionsConfig.java

@@ -26,6 +26,7 @@
 import java.util.stream.Stream;
 
 import io.aklivity.zilla.runtime.binding.mqtt.internal.config.MqttVersion;
+import io.aklivity.zilla.runtime.engine.config.ModelConfig;
 import io.aklivity.zilla.runtime.engine.config.OptionsConfig;
 
 public class MqttOptionsConfig extends OptionsConfig
@@ -50,18 +51,24 @@ public MqttOptionsConfig(
         List<MqttTopicConfig> topics,
         List<MqttVersion> versions)
     {
-        super(topics != null && !topics.isEmpty()
+        super(resolveModels(topics), List.of());
+        this.authorization = authorization;
+        this.topics = topics;
+        this.versions = versions;
+    }
+
+    private static List<ModelConfig> resolveModels(
+        List<MqttTopicConfig> topics)
+    {
+        return topics != null && !topics.isEmpty()
             ? topics.stream()
             .flatMap(topic -> Stream.concat(
-                Stream.of(topic.content),
+                    Stream.of(topic.content),
                     Optional.ofNullable(topic.userProperties).orElseGet(Collections::emptyList).stream()
-                    .flatMap(p -> Stream.of(p.value))
-                    .filter(Objects::nonNull))
+                        .flatMap(p -> Stream.of(p.value))
+                        .filter(Objects::nonNull))
                 .filter(Objects::nonNull))
             .collect(Collectors.toList())
-            : emptyList());
-        this.authorization = authorization;
-        this.topics = topics;
-        this.versions = versions;
+            : emptyList();
     }
 }

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

@@ -64,7 +64,6 @@
 import io.aklivity.zilla.runtime.engine.internal.registry.EngineManager;
 import io.aklivity.zilla.runtime.engine.internal.registry.EngineWorker;
 import io.aklivity.zilla.runtime.engine.internal.types.event.EventFW;
-import io.aklivity.zilla.runtime.engine.internal.watcher.EngineConfigWatcher;
 import io.aklivity.zilla.runtime.engine.metrics.Collector;
 import io.aklivity.zilla.runtime.engine.metrics.MetricGroup;
 import io.aklivity.zilla.runtime.engine.model.Model;
@@ -87,7 +86,6 @@ public final class Engine implements Collector, AutoCloseable
     private final boolean readonly;
     private final EngineConfiguration config;
     private final EngineManager manager;
-    private final EngineConfigWatcher watcher;
 
     Engine(
         EngineConfiguration config,
@@ -148,14 +146,13 @@ public final class Engine implements Collector, AutoCloseable
         }
         this.tuning = tuning;
 
-        this.watcher = new EngineConfigWatcher();
         List<EngineWorker> workers = new ArrayList<>(workerCount);
         for (int workerIndex = 0; workerIndex < workerCount; workerIndex++)
         {
             EngineWorker worker =
                 new EngineWorker(config, tasks, labels, errorHandler, tuning::affinity, bindings, exporters,
                     guards, vaults, catalogs, models, metricGroups, this, this::supplyEventReader,
-                    eventFormatterFactory, watcher, workerIndex, readonly, this::process);
+                    eventFormatterFactory, workerIndex, readonly, this::process);
             workers.add(worker);
         }
         this.workers = workers;
@@ -183,6 +180,7 @@ public final class Engine implements Collector, AutoCloseable
             .collect(Collectors.toMap(g -> g.name(), g -> g));
 
         ResourceResolver resources = new ResourceResolver(catalogs, vaults);
+        this.configURL = config.configURL();
         EngineManager manager = new EngineManager(
             schemaTypes,
             bindingsByType::get,
@@ -196,13 +194,9 @@ public final class Engine implements Collector, AutoCloseable
             context,
             config,
             extensions,
+            this.configURL,
             this::readURL,
-            resources,
-            watcher);
-
-        this.configURL = config.configURL();
-        this.watcher.initialize(configURL, manager::reconfigure, manager::reloadNamespacesWithChangedResources,
-            this::readURL);
+            resources);
 
         this.bindings = bindings;
         this.tasks = tasks;
@@ -235,10 +229,10 @@ public void start() throws Exception
             worker.doStart();
         }
 
+        // ignore the config file in read-only mode; no config will be read so no namespaces, bindings, etc. will be attached
         if (!readonly)
         {
-            // ignore the config file in read-only mode; no config will be read so no namespaces, bindings, etc will be attached
-            watcher.startWatchingConfig();
+            manager.startWatcher();
         }
     }
 
@@ -252,7 +246,7 @@ public void close() throws Exception
 
         final List<Throwable> errors = new ArrayList<>();
 
-        watcher.close();
+        manager.closeWatcher();
 
         for (EngineWorker worker : workers)
         {

runtime/engine/src/main/java/io/aklivity/zilla/runtime/engine/catalog/CatalogHandler.java

@@ -15,8 +15,6 @@
  */
 package io.aklivity.zilla.runtime.engine.catalog;
 
-import java.util.List;
-
 import org.agrona.DirectBuffer;
 
 import io.aklivity.zilla.runtime.engine.model.function.ValueConsumer;
@@ -116,9 +114,4 @@ default int decodePadding(
     {
         return 0;
     }
-
-    default List<String> resources()
-    {
-        return List.of();
-    }
 }

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

@@ -55,6 +55,7 @@
 import io.aklivity.zilla.runtime.engine.EngineConfiguration;
 import io.aklivity.zilla.runtime.engine.internal.config.NamespaceAdapter;
 import io.aklivity.zilla.runtime.engine.internal.config.schema.UniquePropertyKeysSchema;
+import io.aklivity.zilla.runtime.engine.internal.watcher.EngineConfigWatcher;
 import io.aklivity.zilla.runtime.engine.resolver.Resolver;
 
 public final class EngineConfigReader
@@ -64,6 +65,7 @@ public final class EngineConfigReader
     private final Resolver expressions;
     private final Collection<URL> schemaTypes;
     private final ResourceResolver resources;
+    private final EngineConfigWatcher watcher;
     private final Consumer<String> logger;
     private final MessageDigest md5;
 
@@ -73,13 +75,15 @@ public EngineConfigReader(
         Resolver expressions,
         Collection<URL> schemaTypes,
         ResourceResolver resources,
+        EngineConfigWatcher watcher,
         Consumer<String> logger)
     {
         this.config = config;
         this.context = context;
         this.expressions = expressions;
         this.schemaTypes = schemaTypes;
         this.resources = resources;
+        this.watcher = watcher;
         this.logger = logger;
         this.md5 = initMessageDigest("MD5");
     }
@@ -182,7 +186,7 @@ public EngineConfig read(
                 reader.skip(configsAt.get(i));
                 NamespaceConfig namespace = jsonb.fromJson(reader, NamespaceConfig.class);
                 namespace.hash = hashes.get(i);
-                System.out.println(resources.resolve(namespace)); // TODO: Ati
+                watcher.addResources(resources.resolve(namespace), namespace.name);
                 builder.namespace(namespace);
 
                 if (!errors.isEmpty())

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

@@ -21,15 +21,18 @@
 public class OptionsConfig
 {
     public final List<ModelConfig> models;
+    public final List<String> resources;
 
     public OptionsConfig()
     {
-        this(Collections.emptyList());
+        this(Collections.emptyList(), Collections.emptyList());
     }
 
     public OptionsConfig(
-        List<ModelConfig> models)
+        List<ModelConfig> models,
+        List<String> resources)
     {
         this.models = models;
+        this.resources = resources;
     }
 }