Add sse payload validation

runtime/binding-sse/src/main/java/io/aklivity/zilla/runtime/binding/sse/config/SseOptionsConfig.java

@@ -15,15 +15,33 @@
  */
 package io.aklivity.zilla.runtime.binding.sse.config;
 
+import java.util.List;
+import java.util.function.Function;
+
 import io.aklivity.zilla.runtime.engine.config.OptionsConfig;
 
 public final class SseOptionsConfig extends OptionsConfig
 {
     public final int retry;
+    public final List<SsePathConfig> paths;
+
+
+    public static SseOptionsConfigBuilder<SseOptionsConfig> builder()
+    {
+        return new SseOptionsConfigBuilder<>(SseOptionsConfig.class::cast);
+    }
+
+    public static <T> SseOptionsConfigBuilder<T> builder(
+        Function<OptionsConfig, T> mapper)
+    {
+        return new SseOptionsConfigBuilder<>(mapper);
+    }
 
-    public SseOptionsConfig(
-        int retry)
+    SseOptionsConfig(
+        int retry,
+        List<SsePathConfig> paths)
     {
         this.retry = retry;
+        this.paths = paths;
     }
 }

runtime/binding-sse/src/main/java/io/aklivity/zilla/runtime/binding/sse/config/SseOptionsConfigBuilder.java

@@ -0,0 +1,86 @@
+/*
+ * Copyright 2021-2023 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.binding.sse.config;
+
+import java.util.LinkedList;
+import java.util.List;
+import java.util.function.Function;
+
+import io.aklivity.zilla.runtime.engine.config.ConfigBuilder;
+import io.aklivity.zilla.runtime.engine.config.OptionsConfig;
+
+public class SseOptionsConfigBuilder<T> extends ConfigBuilder<T, SseOptionsConfigBuilder<T>>
+{
+    private final Function<OptionsConfig, T> mapper;
+
+    private int retry;
+    private List<SsePathConfig> paths;
+
+    SseOptionsConfigBuilder(
+        Function<OptionsConfig, T> mapper)
+    {
+        this.mapper = mapper;
+    }
+
+    @Override
+    @SuppressWarnings("unchecked")
+    protected Class<SseOptionsConfigBuilder<T>> thisType()
+    {
+        return (Class<SseOptionsConfigBuilder<T>>) getClass();
+    }
+
+    public SseOptionsConfigBuilder<T> retry(
+        int retry)
+    {
+
+        this.retry = retry;
+        return this;
+    }
+
+    public SseOptionsConfigBuilder<T> paths(
+        List<SsePathConfig> paths)
+    {
+        if (paths == null)
+        {
+            paths = new LinkedList<>();
+        }
+        this.paths = paths;
+        return this;
+    }
+
+    public SseOptionsConfigBuilder<T> path(
+        SsePathConfig path)
+    {
+        if (this.paths == null)
+        {
+            this.paths = new LinkedList<>();
+        }
+        this.paths.add(path);
+        return this;
+    }
+
+    public SsePathConfigBuilder<SseOptionsConfigBuilder<T>> path()
+    {
+        return new SsePathConfigBuilder<>(this::path);
+    }
+
+    @Override
+    public T build()
+    {
+        return mapper.apply(new SseOptionsConfig(retry, paths));
+    }
+
+}

runtime/binding-sse/src/main/java/io/aklivity/zilla/runtime/binding/sse/config/SsePathConfig.java

@@ -0,0 +1,40 @@
+/*
+ * Copyright 2021-2023 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.binding.sse.config;
+
+import static java.util.function.Function.identity;
+
+import io.aklivity.zilla.runtime.engine.config.ModelConfig;
+
+public class SsePathConfig
+{
+    public final String path;
+    public final ModelConfig content;
+
+
+    SsePathConfig(
+        String path,
+        ModelConfig content)
+    {
+        this.path = path;
+        this.content = content;
+    }
+
+    public static SsePathConfigBuilder<SsePathConfig> builder()
+    {
+        return new SsePathConfigBuilder<>(identity());
+    }
+}

runtime/binding-sse/src/main/java/io/aklivity/zilla/runtime/binding/sse/config/SsePathConfigBuilder.java

@@ -0,0 +1,68 @@
+/*
+ * Copyright 2021-2023 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.binding.sse.config;
+
+import java.util.function.Function;
+
+import io.aklivity.zilla.runtime.engine.config.ConfigBuilder;
+import io.aklivity.zilla.runtime.engine.config.ModelConfig;
+
+public class SsePathConfigBuilder<T> extends ConfigBuilder<T, SsePathConfigBuilder<T>>
+{
+    private final Function<SsePathConfig, T> mapper;
+
+    private String path;
+    private ModelConfig content;
+
+    SsePathConfigBuilder(
+        Function<SsePathConfig, T> mapper)
+    {
+        this.mapper = mapper;
+    }
+
+    @Override
+    @SuppressWarnings("unchecked")
+    protected Class<SsePathConfigBuilder<T>> thisType()
+    {
+        return (Class<SsePathConfigBuilder<T>>) getClass();
+    }
+
+    public SsePathConfigBuilder<T> path(
+        String path)
+    {
+        this.path = path;
+        return this;
+    }
+
+    public SsePathConfigBuilder<T> content(
+        ModelConfig content)
+    {
+        this.content = content;
+        return this;
+    }
+
+    public <C extends ConfigBuilder<SsePathConfigBuilder<T>, C>> C content(
+        Function<Function<ModelConfig, SsePathConfigBuilder<T>>, C> content)
+    {
+        return content.apply(this::content);
+    }
+
+    @Override
+    public T build()
+    {
+        return mapper.apply(new SsePathConfig(path, content));
+    }
+}

runtime/binding-sse/src/main/java/io/aklivity/zilla/runtime/binding/sse/internal/config/SseBindingConfig.java

@@ -18,21 +18,27 @@
 import static io.aklivity.zilla.runtime.binding.sse.internal.config.SseOptionsConfigAdapter.RETRY_DEFAULT;
 import static java.util.stream.Collectors.toList;
 
+import java.util.HashMap;
 import java.util.List;
+import java.util.Map;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
 
 import io.aklivity.zilla.runtime.binding.sse.config.SseOptionsConfig;
 import io.aklivity.zilla.runtime.engine.config.BindingConfig;
 import io.aklivity.zilla.runtime.engine.config.KindConfig;
+import io.aklivity.zilla.runtime.engine.config.ModelConfig;
 
 public final class SseBindingConfig
 {
-    private static final SseOptionsConfig DEFAULT_OPTIONS = new SseOptionsConfig(RETRY_DEFAULT);
+    private static final SseOptionsConfig DEFAULT_OPTIONS = SseOptionsConfig.builder().retry(RETRY_DEFAULT).build();
 
     public final long id;
     public final String name;
     public final SseOptionsConfig options;
     public final KindConfig kind;
     public final List<SseRouteConfig> routes;
+    public final Map<Matcher, ModelConfig> paths;
 
     public SseBindingConfig(
         BindingConfig binding)
@@ -42,6 +48,12 @@ public SseBindingConfig(
         this.kind = binding.kind;
         this.options = binding.options instanceof SseOptionsConfig ? (SseOptionsConfig) binding.options : DEFAULT_OPTIONS;
         this.routes = binding.routes.stream().map(SseRouteConfig::new).collect(toList());
+        this.paths = new HashMap<>();
+        if (options.paths != null)
+        {
+            options.paths.forEach(p ->
+                paths.put(Pattern.compile(p.path).matcher(""), p.content));
+        }
     }
 
     public SseRouteConfig resolve(
@@ -53,4 +65,24 @@ public SseRouteConfig resolve(
             .findFirst()
             .orElse(null);
     }
+
+    public ModelConfig supplyModelConfig(
+        String path)
+    {
+        ModelConfig config = null;
+        if (paths != null)
+        {
+            for (Map.Entry<Matcher, ModelConfig> e : paths.entrySet())
+            {
+                final Matcher matcher = e.getKey();
+                matcher.reset(path);
+                if (matcher.find())
+                {
+                    config = e.getValue();
+                    break;
+                }
+            }
+        }
+        return config;
+    }
 }

runtime/binding-sse/src/main/java/io/aklivity/zilla/runtime/binding/sse/internal/config/SseOptionsConfigAdapter.java

@@ -15,21 +15,31 @@
  */
 package io.aklivity.zilla.runtime.binding.sse.internal.config;
 
+import java.util.List;
+import java.util.stream.Collectors;
+
 import jakarta.json.Json;
+import jakarta.json.JsonArrayBuilder;
 import jakarta.json.JsonObject;
 import jakarta.json.JsonObjectBuilder;
 import jakarta.json.bind.adapter.JsonbAdapter;
 
 import io.aklivity.zilla.runtime.binding.sse.config.SseOptionsConfig;
+import io.aklivity.zilla.runtime.binding.sse.config.SseOptionsConfigBuilder;
+import io.aklivity.zilla.runtime.binding.sse.config.SsePathConfig;
 import io.aklivity.zilla.runtime.binding.sse.internal.SseBinding;
 import io.aklivity.zilla.runtime.engine.config.OptionsConfig;
 import io.aklivity.zilla.runtime.engine.config.OptionsConfigAdapterSpi;
 
 public final class SseOptionsConfigAdapter implements OptionsConfigAdapterSpi, JsonbAdapter<OptionsConfig, JsonObject>
 {
     private static final String RETRY_NAME = "retry";
+    private static final String PATHS_NAME = "paths";
     public static final int RETRY_DEFAULT = 2000;
 
+
+    private final SsePathConfigAdapter ssePath = new SsePathConfigAdapter();
+
     @Override
     public Kind kind()
     {
@@ -55,17 +65,41 @@ public JsonObject adaptToJson(
             object.add(RETRY_NAME, sseOptions.retry);
         }
 
+        if (sseOptions.paths != null)
+        {
+            JsonArrayBuilder paths = Json.createArrayBuilder();
+            sseOptions.paths.stream()
+                .map(ssePath::adaptToJson)
+                .forEach(paths::add);
+            object.add(PATHS_NAME, paths);
+        }
+
         return object.build();
     }
 
     @Override
     public OptionsConfig adaptFromJson(
         JsonObject object)
     {
-        int retry = object.containsKey(RETRY_NAME)
-                ? object.getInt(RETRY_NAME)
-                : SseOptionsConfigAdapter.RETRY_DEFAULT;
+        SseOptionsConfigBuilder<SseOptionsConfig> sseOptions = SseOptionsConfig.builder();
+
+        if (object.containsKey(RETRY_NAME))
+        {
+            sseOptions.retry(object.getInt(RETRY_NAME));
+        }
+        else
+        {
+            sseOptions.retry(SseOptionsConfigAdapter.RETRY_DEFAULT);
+        }
+
+        if (object.containsKey(PATHS_NAME))
+        {
+            List<SsePathConfig> paths = object.getJsonArray(PATHS_NAME).stream()
+                .map(item -> ssePath.adaptFromJson((JsonObject) item))
+                .collect(Collectors.toList());
+            sseOptions.paths(paths);
+        }
 
-        return new SseOptionsConfig(retry);
+        return sseOptions.build();
     }
 }