Include validation in the asyncapi.http.proxy generator

incubator/command-generate/src/main/java/io/aklivity/zilla/runtime/command/generate/internal/airline/ConfigGenerator.java

@@ -14,13 +14,138 @@
  */
 package io.aklivity.zilla.runtime.command.generate.internal.airline;
 
+import static com.fasterxml.jackson.dataformat.yaml.YAMLGenerator.Feature.MINIMIZE_QUOTES;
+import static com.fasterxml.jackson.dataformat.yaml.YAMLGenerator.Feature.WRITE_DOC_START_MARKER;
+import static org.agrona.LangUtil.rethrowUnchecked;
+
 import java.util.List;
+import java.util.Map;
+import java.util.regex.Matcher;
 import java.util.regex.Pattern;
 
+import jakarta.json.bind.Jsonb;
+import jakarta.json.bind.JsonbBuilder;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.dataformat.yaml.YAMLMapper;
+
+import io.aklivity.zilla.runtime.catalog.inline.config.InlineOptionsConfig;
+import io.aklivity.zilla.runtime.catalog.inline.config.InlineSchemaConfigBuilder;
+import io.aklivity.zilla.runtime.command.generate.internal.asyncapi.model.AsyncApi;
+import io.aklivity.zilla.runtime.command.generate.internal.asyncapi.model.Message;
+import io.aklivity.zilla.runtime.command.generate.internal.asyncapi.model.Schema;
+import io.aklivity.zilla.runtime.command.generate.internal.asyncapi.view.MessageView;
+import io.aklivity.zilla.runtime.engine.config.NamespaceConfig;
+import io.aklivity.zilla.runtime.engine.config.NamespaceConfigBuilder;
+import io.aklivity.zilla.runtime.engine.config.ValidatorConfig;
+import io.aklivity.zilla.runtime.validator.core.config.IntegerValidatorConfig;
+import io.aklivity.zilla.runtime.validator.core.config.StringValidatorConfig;
+
 public abstract class ConfigGenerator
 {
+    protected static final String INLINE_CATALOG_NAME = "catalog0";
+    protected static final String INLINE_CATALOG_TYPE = "inline";
+    protected static final String APPLICATION_JSON = "application/json";
+    protected static final String VERSION_LATEST = "latest";
+    protected static final Matcher JSON_CONTENT_TYPE = Pattern.compile("^application/(?:.+\\+)?json$").matcher("");
+
+    protected final Map<String, ValidatorConfig> validators = Map.of(
+        "string", StringValidatorConfig.builder().build(),
+        "integer", IntegerValidatorConfig.builder().build()
+    );
+
+    protected AsyncApi asyncApi;
+
     public abstract String generate();
 
+    protected boolean jsonContentType()
+    {
+        boolean result = false;
+        String contentType = resolveContentType();
+        if (contentType != null)
+        {
+            result = JSON_CONTENT_TYPE.reset(contentType).matches();
+        }
+        return result;
+    }
+
+    private String resolveContentType()
+    {
+        String contentType = null;
+        if (asyncApi.components != null && asyncApi.components.messages != null && !asyncApi.components.messages.isEmpty())
+        {
+            Message firstMessage = asyncApi.components.messages.entrySet().stream().findFirst().get().getValue();
+            contentType = MessageView.of(asyncApi.components.messages, firstMessage).contentType();
+        }
+        return contentType;
+    }
+
+    protected NamespaceConfigBuilder<NamespaceConfig> injectCatalog(
+        NamespaceConfigBuilder<NamespaceConfig> namespace)
+    {
+        if (asyncApi.components != null && asyncApi.components.schemas != null && !asyncApi.components.schemas.isEmpty())
+        {
+            namespace
+                .catalog()
+                    .name(INLINE_CATALOG_NAME)
+                    .type(INLINE_CATALOG_TYPE)
+                    .options(InlineOptionsConfig::builder)
+                        .subjects()
+                            .inject(this::injectSubjects)
+                            .build()
+                        .build()
+                    .build();
+
+        }
+        return namespace;
+    }
+
+    private <C> InlineSchemaConfigBuilder<C> injectSubjects(
+        InlineSchemaConfigBuilder<C> subjects)
+    {
+        try (Jsonb jsonb = JsonbBuilder.create())
+        {
+            YAMLMapper yaml = YAMLMapper.builder()
+                .disable(WRITE_DOC_START_MARKER)
+                .enable(MINIMIZE_QUOTES)
+                .build();
+            for (Map.Entry<String, Schema> entry : asyncApi.components.schemas.entrySet())
+            {
+                subjects
+                    .subject(entry.getKey())
+                        .version(VERSION_LATEST)
+                        .schema(writeSchemaYaml(jsonb, yaml, entry.getValue()))
+                        .build();
+            }
+        }
+        catch (Exception ex)
+        {
+            rethrowUnchecked(ex);
+        }
+        return subjects;
+    }
+
+    private static String writeSchemaYaml(
+        Jsonb jsonb,
+        YAMLMapper yaml,
+        Schema schema)
+    {
+        String result = null;
+        try
+        {
+            String schemaJson = jsonb.toJson(schema);
+            JsonNode json = new ObjectMapper().readTree(schemaJson);
+            result = yaml.writeValueAsString(json);
+        }
+        catch (JsonProcessingException ex)
+        {
+            rethrowUnchecked(ex);
+        }
+        return result;
+    }
+
     protected final String unquoteEnvVars(
         String yaml,
         List<String> unquotedEnvVars)

incubator/command-generate/src/main/java/io/aklivity/zilla/runtime/command/generate/internal/asyncapi/http/proxy/AsyncApiHttpProxyConfigGenerator.java

@@ -32,9 +32,13 @@
 import jakarta.json.bind.Jsonb;
 import jakarta.json.bind.JsonbBuilder;
 
+import org.apache.commons.collections4.MapUtils;
+
 import io.aklivity.zilla.runtime.binding.http.config.HttpConditionConfig;
 import io.aklivity.zilla.runtime.binding.http.config.HttpOptionsConfig;
 import io.aklivity.zilla.runtime.binding.http.config.HttpOptionsConfigBuilder;
+import io.aklivity.zilla.runtime.binding.http.config.HttpRequestConfig.Method;
+import io.aklivity.zilla.runtime.binding.http.config.HttpRequestConfigBuilder;
 import io.aklivity.zilla.runtime.binding.tcp.config.TcpConditionConfig;
 import io.aklivity.zilla.runtime.binding.tcp.config.TcpOptionsConfig;
 import io.aklivity.zilla.runtime.binding.tls.config.TlsOptionsConfig;
@@ -43,23 +47,27 @@
 import io.aklivity.zilla.runtime.command.generate.internal.asyncapi.model.Item;
 import io.aklivity.zilla.runtime.command.generate.internal.asyncapi.model.Message;
 import io.aklivity.zilla.runtime.command.generate.internal.asyncapi.model.Operation;
+import io.aklivity.zilla.runtime.command.generate.internal.asyncapi.model.Parameter;
 import io.aklivity.zilla.runtime.command.generate.internal.asyncapi.model.Server;
 import io.aklivity.zilla.runtime.command.generate.internal.asyncapi.view.ChannelView;
+import io.aklivity.zilla.runtime.command.generate.internal.asyncapi.view.MessageView;
 import io.aklivity.zilla.runtime.command.generate.internal.asyncapi.view.ServerView;
 import io.aklivity.zilla.runtime.engine.config.BindingConfigBuilder;
+import io.aklivity.zilla.runtime.engine.config.CatalogedConfigBuilder;
 import io.aklivity.zilla.runtime.engine.config.ConfigWriter;
 import io.aklivity.zilla.runtime.engine.config.GuardedConfigBuilder;
 import io.aklivity.zilla.runtime.engine.config.NamespaceConfig;
 import io.aklivity.zilla.runtime.engine.config.NamespaceConfigBuilder;
 import io.aklivity.zilla.runtime.engine.config.RouteConfigBuilder;
+import io.aklivity.zilla.runtime.engine.config.ValidatorConfig;
 import io.aklivity.zilla.runtime.guard.jwt.config.JwtOptionsConfig;
+import io.aklivity.zilla.runtime.validator.json.config.JsonValidatorConfig;
 import io.aklivity.zilla.runtime.vault.filesystem.config.FileSystemOptionsConfig;
 
 public class AsyncApiHttpProxyConfigGenerator extends ConfigGenerator
 {
     private final InputStream input;
 
-    private AsyncApi asyncApi;
     private int[] allPorts;
     private int[] httpPorts;
     private int[] httpsPorts;
@@ -172,18 +180,20 @@ private String resolveAuthorizationHeader()
     {
         requireNonNull(asyncApi);
         requireNonNull(asyncApi.components);
-        requireNonNull(asyncApi.components.messages);
         String result = null;
-        for (Map.Entry<String, Message> entry: asyncApi.components.messages.entrySet())
+        if (asyncApi.components.messages != null)
         {
-            Message message = entry.getValue();
-            if (message.headers != null && message.headers.properties != null)
+            for (Map.Entry<String, Message> entry : asyncApi.components.messages.entrySet())
             {
-                Item authorization = message.headers.properties.get("authorization");
-                if (authorization != null)
+                Message message = entry.getValue();
+                if (message.headers != null && message.headers.properties != null)
                 {
-                    result = authorization.description;
-                    break;
+                    Item authorization = message.headers.properties.get("authorization");
+                    if (authorization != null)
+                    {
+                        result = authorization.description;
+                        break;
+                    }
                 }
             }
         }
@@ -215,6 +225,7 @@ private NamespaceConfig createNamespace()
                         .policy(CROSS_ORIGIN)
                         .build()
                     .inject(this::injectHttpServerOptions)
+                    .inject(this::injectHttpServerRequests)
                     .build()
                 .inject(this::injectHttpServerRoutes)
                 .build()
@@ -236,6 +247,7 @@ private NamespaceConfig createNamespace()
                 .build()
             .inject(this::injectGuard)
             .inject(this::injectVaults)
+            .inject(this::injectCatalog)
             .build();
     }
 
@@ -293,8 +305,8 @@ private NamespaceConfigBuilder<NamespaceConfig> injectTlsServer(
         return namespace;
     }
 
-    private HttpOptionsConfigBuilder<BindingConfigBuilder<NamespaceConfigBuilder<NamespaceConfig>>> injectHttpServerOptions(
-        HttpOptionsConfigBuilder<BindingConfigBuilder<NamespaceConfigBuilder<NamespaceConfig>>> options)
+    private <C> HttpOptionsConfigBuilder<C> injectHttpServerOptions(
+        HttpOptionsConfigBuilder<C> options)
     {
         if (isJwtEnabled)
         {
@@ -312,6 +324,93 @@ private HttpOptionsConfigBuilder<BindingConfigBuilder<NamespaceConfigBuilder<Nam
         return options;
     }
 
+    private <C> HttpOptionsConfigBuilder<C> injectHttpServerRequests(
+        HttpOptionsConfigBuilder<C> options)
+    {
+        for (String name : asyncApi.operations.keySet())
+        {
+            Operation operation = asyncApi.operations.get(name);
+            ChannelView channel = ChannelView.of(asyncApi.channels, operation.channel);
+            String path = channel.address();
+            Method method = Method.valueOf(operation.bindings.get("http").method);
+            if (MapUtils.isNotEmpty(channel.messages()) || MapUtils.isNotEmpty(channel.parameters()))
+            {
+                options
+                    .request()
+                        .path(path)
+                        .method(method)
+                        .inject(request -> injectContent(request, channel.messages()))
+                        .inject(request -> injectPathParams(request, channel.parameters()))
+                        .build();
+            }
+        }
+        return options;
+    }
+
+    private <C> HttpRequestConfigBuilder<C> injectContent(
+        HttpRequestConfigBuilder<C> request,
+        Map<String, Message> messages)
+    {
+        if (messages != null)
+        {
+            if (jsonContentType())
+            {
+                request.
+                    content(JsonValidatorConfig::builder)
+                        .catalog()
+                            .name(INLINE_CATALOG_NAME)
+                            .inject(catalog -> injectSchemas(catalog, messages))
+                            .build()
+                        .build();
+            }
+        }
+        return request;
+    }
+
+    private <C> CatalogedConfigBuilder<C> injectSchemas(
+        CatalogedConfigBuilder<C> catalog,
+        Map<String, Message> messages)
+    {
+        for (String name : messages.keySet())
+        {
+            MessageView message = MessageView.of(asyncApi.components.messages, messages.get(name));
+            String subject = message.refKey() != null ? message.refKey() : name;
+            catalog
+                .schema()
+                    .subject(subject)
+                    .build()
+                .build();
+        }
+        return catalog;
+    }
+
+    private <C> HttpRequestConfigBuilder<C> injectPathParams(
+        HttpRequestConfigBuilder<C> request,
+        Map<String, Parameter> parameters)
+    {
+        if (parameters != null)
+        {
+            for (String name : parameters.keySet())
+            {
+                Parameter parameter = parameters.get(name);
+                if (parameter.schema != null && parameter.schema.type != null)
+                {
+                    ValidatorConfig validator = validators.get(parameter.schema.type);
+                    if (validator != null)
+                    {
+                        request
+                            .pathParam()
+                                .name(name)
+                                .validator(validator)
+                                .build();
+                    }
+                }
+            }
+        }
+        return request;
+    }
+
+
     private BindingConfigBuilder<NamespaceConfigBuilder<NamespaceConfig>> injectHttpServerRoutes(
         BindingConfigBuilder<NamespaceConfigBuilder<NamespaceConfig>> binding)
     {

incubator/command-generate/src/main/java/io/aklivity/zilla/runtime/command/generate/internal/asyncapi/model/Channel.java

@@ -22,6 +22,7 @@ public class Channel
 {
     public String address;
     public LinkedHashMap<String, Message> messages;
+    public LinkedHashMap<String, Parameter> parameters;
 
     @JsonbProperty("$ref")
     public String ref;

incubator/command-generate/src/main/java/io/aklivity/zilla/runtime/command/generate/internal/asyncapi/model/Parameter.java

@@ -0,0 +1,20 @@
+/*
+ * Copyright 2021-2023 Aklivity Inc
+ *
+ * Licensed under the Aklivity Community License (the "License"); you may not use
+ * this file except in compliance with the License.  You may obtain a copy of the
+ * License at
+ *
+ *   https://www.aklivity.io/aklivity-community-license/
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES 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.command.generate.internal.asyncapi.model;
+
+public class Parameter
+{
+    public Schema schema;
+}

incubator/command-generate/src/main/java/io/aklivity/zilla/runtime/command/generate/internal/asyncapi/model/Schema.java

@@ -17,8 +17,16 @@
 import java.util.List;
 import java.util.Map;
 
+import jakarta.json.bind.annotation.JsonbPropertyOrder;
+
+@JsonbPropertyOrder({
+    "type",
+    "properties",
+    "required"
+})
 public class Schema
 {
+    public String type;
     public Map<String, Item> properties;
     public List<String> required;
 }