Support remote zilla configuration with change detection

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/pom.xml

@@ -26,7 +26,7 @@
   <properties>
     <maven.compiler.source>11</maven.compiler.source>
     <maven.compiler.target>11</maven.compiler.target>
-    <jacoco.coverage.ratio>0.77</jacoco.coverage.ratio>
+    <jacoco.coverage.ratio>0.74</jacoco.coverage.ratio>
     <jacoco.missed.count>5</jacoco.missed.count>
   </properties>
 

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

@@ -16,21 +16,14 @@
 package io.aklivity.zilla.runtime.engine;
 
 import static io.aklivity.zilla.runtime.engine.internal.layouts.metrics.HistogramsLayout.BUCKETS;
-import static java.net.http.HttpClient.Redirect.NORMAL;
-import static java.net.http.HttpClient.Version.HTTP_2;
 import static java.nio.charset.StandardCharsets.UTF_8;
 import static java.util.concurrent.Executors.newFixedThreadPool;
 import static java.util.stream.Collectors.toList;
 import static org.agrona.LangUtil.rethrowUnchecked;
 
-import java.io.IOException;
 import java.io.InputStream;
-import java.net.URISyntaxException;
 import java.net.URL;
 import java.net.URLConnection;
-import java.net.http.HttpClient;
-import java.net.http.HttpRequest;
-import java.net.http.HttpResponse;
 import java.util.ArrayList;
 import java.util.Collection;
 import java.util.List;
@@ -40,11 +33,9 @@
 import java.util.ServiceLoader.Provider;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
-import java.util.concurrent.Future;
 import java.util.concurrent.ThreadFactory;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.function.Consumer;
-import java.util.function.Function;
 import java.util.function.IntFunction;
 import java.util.function.LongConsumer;
 import java.util.function.LongSupplier;
@@ -60,6 +51,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.ResourceResolver;
 import io.aklivity.zilla.runtime.engine.event.EventFormatterFactory;
 import io.aklivity.zilla.runtime.engine.exporter.Exporter;
 import io.aklivity.zilla.runtime.engine.ext.EngineExtContext;
@@ -71,9 +63,6 @@
 import io.aklivity.zilla.runtime.engine.internal.layouts.EventsLayout;
 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.registry.FileWatcherTask;
-import io.aklivity.zilla.runtime.engine.internal.registry.HttpWatcherTask;
-import io.aklivity.zilla.runtime.engine.internal.registry.WatcherTask;
 import io.aklivity.zilla.runtime.engine.internal.types.event.EventFW;
 import io.aklivity.zilla.runtime.engine.metrics.Collector;
 import io.aklivity.zilla.runtime.engine.metrics.MetricGroup;
@@ -92,15 +81,12 @@ public final class Engine implements Collector, AutoCloseable
     private final AtomicInteger nextTaskId;
     private final ThreadFactory factory;
 
-    private final WatcherTask watcherTask;
     private final URL configURL;
     private final List<EngineWorker> workers;
     private final boolean readonly;
     private final EngineConfiguration config;
     private final EngineManager manager;
 
-    private Future<Void> watcherTaskRef;
-
     Engine(
         EngineConfiguration config,
         Collection<Binding> bindings,
@@ -193,6 +179,8 @@ public final class Engine implements Collector, AutoCloseable
         final Map<String, Guard> guardsByType = guards.stream()
             .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,
@@ -206,23 +194,9 @@ public final class Engine implements Collector, AutoCloseable
             context,
             config,
             extensions,
-            this::readURL);
-
-        this.configURL = config.configURL();
-        String protocol = configURL.getProtocol();
-        if ("file".equals(protocol) || "jar".equals(protocol))
-        {
-            Function<String, String> watcherReadURL = l -> readURL(configURL, l);
-            this.watcherTask = new FileWatcherTask(manager::reconfigure, watcherReadURL);
-        }
-        else if ("http".equals(protocol) || "https".equals(protocol))
-        {
-            this.watcherTask = new HttpWatcherTask(manager::reconfigure, config.configPollIntervalSeconds());
-        }
-        else
-        {
-            throw new UnsupportedOperationException();
-        }
+            this.configURL,
+            this::readURL,
+            resources);
 
         this.bindings = bindings;
         this.tasks = tasks;
@@ -255,11 +229,10 @@ public void start() throws Exception
             worker.doStart();
         }
 
-        watcherTaskRef = watcherTask.submit();
+        // 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
-            watcherTask.watch(configURL).get();
+            manager.startWatcher();
         }
     }
 
@@ -273,8 +246,7 @@ public void close() throws Exception
 
         final List<Throwable> errors = new ArrayList<>();
 
-        watcherTask.close();
-        watcherTaskRef.get();
+        manager.closeWatcher();
 
         for (EngineWorker worker : workers)
         {
@@ -317,46 +289,23 @@ public static EngineBuilder builder()
     }
 
     private String readURL(
-        URL configURL,
         String location)
     {
-        String output = null;
+        String result;
         try
         {
-            final URL fileURL = new URL(configURL, location);
-            if ("http".equals(fileURL.getProtocol()) || "https".equals(fileURL.getProtocol()))
+            URL url = new URL(configURL, location);
+            URLConnection connection = url.openConnection();
+            try (InputStream input = connection.getInputStream())
             {
-                HttpClient client = HttpClient.newBuilder()
-                    .version(HTTP_2)
-                    .followRedirects(NORMAL)
-                    .build();
-
-                HttpRequest request = HttpRequest.newBuilder()
-                    .GET()
-                    .uri(fileURL.toURI())
-                    .build();
-
-                HttpResponse<String> response = client.send(
-                    request,
-                    HttpResponse.BodyHandlers.ofString());
-
-                output = response.body();
-            }
-            else
-            {
-
-                URLConnection connection = fileURL.openConnection();
-                try (InputStream input = connection.getInputStream())
-                {
-                    output = new String(input.readAllBytes(), UTF_8);
-                }
+                result = new String(input.readAllBytes(), UTF_8);
             }
         }
-        catch (IOException | URISyntaxException | InterruptedException ex)
+        catch (Exception ex)
         {
-            output = "";
+            result = "";
         }
-        return output;
+        return result;
     }
 
     private Thread newTaskThread(