Add key filter support in openapi asyncapi mapping

incubator/binding-openapi-asyncapi/src/main/java/io/aklivity/zilla/runtime/binding/openapi/asyncapi/internal/config/OpenapiAsyncCompositeBindingAdapter.java

@@ -17,6 +17,7 @@
 import static io.aklivity.zilla.runtime.engine.config.KindConfig.PROXY;
 import static java.util.stream.Collectors.toList;
 
+import java.util.ArrayList;
 import java.util.List;
 import java.util.Map;
 import java.util.regex.Matcher;
@@ -30,6 +31,7 @@
 import io.aklivity.zilla.runtime.binding.http.kafka.config.HttpKafkaWithConfigBuilder;
 import io.aklivity.zilla.runtime.binding.http.kafka.config.HttpKafkaWithFetchConfig;
 import io.aklivity.zilla.runtime.binding.http.kafka.config.HttpKafkaWithFetchConfigBuilder;
+import io.aklivity.zilla.runtime.binding.http.kafka.config.HttpKafkaWithFetchFilterConfig;
 import io.aklivity.zilla.runtime.binding.http.kafka.config.HttpKafkaWithFetchMergeConfig;
 import io.aklivity.zilla.runtime.binding.http.kafka.config.HttpKafkaWithProduceConfig;
 import io.aklivity.zilla.runtime.binding.openapi.asyncapi.config.OpenapiAsyncapiOptionsConfig;
@@ -49,8 +51,10 @@
 public final class OpenapiAsyncCompositeBindingAdapter implements CompositeBindingAdapterSpi
 {
     private static final Pattern JSON_CONTENT_TYPE = Pattern.compile("^application/(?:.+\\+)?json$");
+    private static final Pattern PARAMETER_PATTERN = Pattern.compile("\\{([^}]+)\\}");
 
     private final Matcher jsonContentType = JSON_CONTENT_TYPE.matcher("");
+    private final Matcher parameters = PARAMETER_PATTERN.matcher("");
 
     @Override
     public String type()
@@ -119,6 +123,13 @@ private <C> BindingConfigBuilder<C> injectHttpKafkaRoutes(
                         final AsyncapiChannelView channel = AsyncapiChannelView
                             .of(asyncapi.channels, asyncapiOperation.channel);
 
+                        List<String> paramNames = new ArrayList<>();
+                        Matcher matcher = parameters.reset(item);
+                        while (matcher.find())
+                        {
+                            paramNames.add(parameters.group(1));
+                        }
+
                         binding
                             .route()
                                 .exit(qname)
@@ -127,7 +138,7 @@ private <C> BindingConfigBuilder<C> injectHttpKafkaRoutes(
                                     .path(item)
                                     .build()
                                 .inject(r -> injectHttpKafkaRouteWith(r, openapi, openapiOperation,
-                                    asyncapiOperation.action, channel.address()))
+                                    asyncapiOperation.action, channel.address(), paramNames))
                                 .build();
                     }
                 }
@@ -142,7 +153,8 @@ private <C> RouteConfigBuilder<C> injectHttpKafkaRouteWith(
         Openapi openapi,
         OpenapiOperation operation,
         String action,
-        String address)
+        String address,
+        List<String> paramNames)
     {
         final HttpKafkaWithConfigBuilder<HttpKafkaWithConfig> newWith = HttpKafkaWithConfig.builder();
 
@@ -151,7 +163,7 @@ private <C> RouteConfigBuilder<C> injectHttpKafkaRouteWith(
         case "receive":
             newWith.fetch(HttpKafkaWithFetchConfig.builder()
                 .topic(address)
-                .inject(with -> this.injectHttpKafkaRouteFetchWith(with, openapi, operation))
+                .inject(with -> this.injectHttpKafkaRouteFetchWith(with, openapi, operation, paramNames))
                 .build());
             break;
         case "send":
@@ -170,7 +182,8 @@ private <C> RouteConfigBuilder<C> injectHttpKafkaRouteWith(
     private <C> HttpKafkaWithFetchConfigBuilder<C> injectHttpKafkaRouteFetchWith(
         HttpKafkaWithFetchConfigBuilder<C> fetch,
         Openapi openapi,
-        OpenapiOperation operation)
+        OpenapiOperation operation,
+        List<String> paramNames)
     {
         merge:
         for (Map.Entry<String, OpenapiResponseByContentType> response : operation.responses.entrySet())
@@ -181,11 +194,20 @@ private <C> HttpKafkaWithFetchConfigBuilder<C> injectHttpKafkaRouteFetchWith(
             {
                 fetch.merged(HttpKafkaWithFetchMergeConfig.builder()
                     .contentType("application/json")
+                    .initial("[]")
+                    .path("/-")
                     .build());
                 break merge;
             }
+        }
 
+        if (!paramNames.isEmpty())
+        {
+            fetch.filters(List.of(HttpKafkaWithFetchFilterConfig.builder()
+                .key(String.format("${params.%s}", paramNames.get(0)))
+                .build()));
         }
+
         return fetch;
     }
 

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

@@ -18,8 +18,6 @@
 import java.util.Optional;
 import java.util.function.Function;
 
-import io.aklivity.zilla.runtime.binding.http.kafka.internal.config.HttpKafkaWithFetchFilterConfig;
-
 public final class HttpKafkaWithFetchConfig
 {
     public final String topic;

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

@@ -17,7 +17,6 @@
 import java.util.List;
 import java.util.function.Function;
 
-import io.aklivity.zilla.runtime.binding.http.kafka.internal.config.HttpKafkaWithFetchFilterConfig;
 import io.aklivity.zilla.runtime.engine.config.ConfigBuilder;
 
 public final class HttpKafkaWithFetchConfigBuilder<T> extends ConfigBuilder<T, HttpKafkaWithFetchConfigBuilder<T>>

runtime/binding-http-kafka/src/main/java/io/aklivity/zilla/runtime/binding/http/kafka/internal/config/HttpKafkaWithFetchFilterConfig.java → runtime/binding-http-kafka/src/main/java/io/aklivity/zilla/runtime/binding/http/kafka/config/HttpKafkaWithFetchFilterConfig.java

@@ -12,10 +12,11 @@
  * 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.binding.http.kafka.internal.config;
+package io.aklivity.zilla.runtime.binding.http.kafka.config;
 
 import java.util.List;
 import java.util.Optional;
+import java.util.function.Function;
 
 public final class HttpKafkaWithFetchFilterConfig
 {
@@ -29,4 +30,15 @@ public HttpKafkaWithFetchFilterConfig(
         this.key = Optional.ofNullable(key);
         this.headers = Optional.ofNullable(headers);
     }
+
+    public static HttpKafkaWithFetchFilterConfigBuilder<HttpKafkaWithFetchFilterConfig> builder()
+    {
+        return new HttpKafkaWithFetchFilterConfigBuilder<>(HttpKafkaWithFetchFilterConfig.class::cast);
+    }
+
+    public static <T> HttpKafkaWithFetchFilterConfigBuilder<T> builder(
+        Function<HttpKafkaWithFetchFilterConfig, T> mapper)
+    {
+        return new HttpKafkaWithFetchFilterConfigBuilder<>(mapper);
+    }
 }

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

@@ -0,0 +1,80 @@
+/*
+ * 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.binding.http.kafka.config;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.function.Function;
+
+import io.aklivity.zilla.runtime.engine.config.ConfigBuilder;
+
+public final class HttpKafkaWithFetchFilterConfigBuilder<T> extends
+    ConfigBuilder<T, HttpKafkaWithFetchFilterConfigBuilder<T>>
+{
+    private final Function<HttpKafkaWithFetchFilterConfig, T> mapper;
+    private String key;
+    private List<HttpKafkaWithFetchFilterHeaderConfig> headers;
+
+
+    HttpKafkaWithFetchFilterConfigBuilder(
+        Function<HttpKafkaWithFetchFilterConfig, T> mapper)
+    {
+        this.mapper = mapper;
+    }
+
+    public HttpKafkaWithFetchFilterConfigBuilder<T> key(
+        String key)
+    {
+        this.key = key;
+        return this;
+    }
+
+    public HttpKafkaWithFetchFilterConfigBuilder<T> headers(
+        List<HttpKafkaWithFetchFilterHeaderConfig> headers)
+    {
+        this.headers = headers;
+        return this;
+    }
+
+    public HttpKafkaWithFetchFilterConfigBuilder<T> header(
+        String name,
+        String value)
+    {
+        if (headers == null)
+        {
+            headers = new ArrayList<>();
+        }
+
+        headers.add(HttpKafkaWithFetchFilterHeaderConfig.builder()
+                        .name(name)
+                        .value(value)
+                        .build());
+
+        return this;
+    }
+
+    @Override
+    @SuppressWarnings("unchecked")
+    protected Class<HttpKafkaWithFetchFilterConfigBuilder<T>> thisType()
+    {
+        return (Class<HttpKafkaWithFetchFilterConfigBuilder<T>>) getClass();
+    }
+
+    @Override
+    public T build()
+    {
+        return mapper.apply(new HttpKafkaWithFetchFilterConfig(key, headers));
+    }
+}

runtime/binding-http-kafka/src/main/java/io/aklivity/zilla/runtime/binding/http/kafka/internal/config/HttpKafkaWithFetchFilterHeaderConfig.java → runtime/binding-http-kafka/src/main/java/io/aklivity/zilla/runtime/binding/http/kafka/config/HttpKafkaWithFetchFilterHeaderConfig.java

@@ -12,7 +12,9 @@
  * 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.binding.http.kafka.internal.config;
+package io.aklivity.zilla.runtime.binding.http.kafka.config;
+
+import java.util.function.Function;
 
 public final class HttpKafkaWithFetchFilterHeaderConfig
 {
@@ -26,4 +28,15 @@ public HttpKafkaWithFetchFilterHeaderConfig(
         this.name = name;
         this.value = value;
     }
+
+    public static HttpKafkaWithFetchFilterHeaderConfigBuilder<HttpKafkaWithFetchFilterHeaderConfig> builder()
+    {
+        return new HttpKafkaWithFetchFilterHeaderConfigBuilder<>(HttpKafkaWithFetchFilterHeaderConfig.class::cast);
+    }
+
+    public static <T> HttpKafkaWithFetchFilterHeaderConfigBuilder<T> builder(
+        Function<HttpKafkaWithFetchFilterHeaderConfig, T> mapper)
+    {
+        return new HttpKafkaWithFetchFilterHeaderConfigBuilder<>(mapper);
+    }
 }

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

@@ -0,0 +1,61 @@
+/*
+ * 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.binding.http.kafka.config;
+
+import java.util.function.Function;
+
+import io.aklivity.zilla.runtime.engine.config.ConfigBuilder;
+
+public final class HttpKafkaWithFetchFilterHeaderConfigBuilder<T> extends
+    ConfigBuilder<T, HttpKafkaWithFetchFilterHeaderConfigBuilder<T>>
+{
+    private final Function<HttpKafkaWithFetchFilterHeaderConfig, T> mapper;
+    private String name;
+    private String value;
+
+
+    HttpKafkaWithFetchFilterHeaderConfigBuilder(
+        Function<HttpKafkaWithFetchFilterHeaderConfig, T> mapper)
+    {
+        this.mapper = mapper;
+    }
+
+    public HttpKafkaWithFetchFilterHeaderConfigBuilder<T> name(
+        String name)
+    {
+        this.name = name;
+        return this;
+    }
+
+    public HttpKafkaWithFetchFilterHeaderConfigBuilder<T> value(
+        String value)
+    {
+        this.value = value;
+        return this;
+    }
+
+    @Override
+    @SuppressWarnings("unchecked")
+    protected Class<HttpKafkaWithFetchFilterHeaderConfigBuilder<T>> thisType()
+    {
+        return (Class<HttpKafkaWithFetchFilterHeaderConfigBuilder<T>>) getClass();
+    }
+
+    @Override
+    public T build()
+    {
+        return mapper.apply(new HttpKafkaWithFetchFilterHeaderConfig(name, value));
+    }
+}

runtime/binding-http-kafka/src/main/java/io/aklivity/zilla/runtime/binding/http/kafka/internal/config/HttpKafkaWithFetchConfigAdapter.java

@@ -26,6 +26,8 @@
 
 import io.aklivity.zilla.runtime.binding.http.kafka.config.HttpKafkaWithConfig;
 import io.aklivity.zilla.runtime.binding.http.kafka.config.HttpKafkaWithFetchConfig;
+import io.aklivity.zilla.runtime.binding.http.kafka.config.HttpKafkaWithFetchFilterConfig;
+import io.aklivity.zilla.runtime.binding.http.kafka.config.HttpKafkaWithFetchFilterHeaderConfig;
 import io.aklivity.zilla.runtime.binding.http.kafka.config.HttpKafkaWithFetchMergeConfig;
 
 public final class HttpKafkaWithFetchConfigAdapter implements JsonbAdapter<HttpKafkaWithConfig, JsonObject>
@@ -142,11 +144,17 @@ public HttpKafkaWithConfig adaptFromJson(
                     for (String newHeaderName : headers.keySet())
                     {
                         String newHeaderValue = headers.getString(newHeaderName);
-                        newHeaders.add(new HttpKafkaWithFetchFilterHeaderConfig(newHeaderName, newHeaderValue));
+                        newHeaders.add(HttpKafkaWithFetchFilterHeaderConfig.builder()
+                            .name(newHeaderName)
+                            .value(newHeaderValue)
+                            .build());
                     }
                 }
 
-                newFilters.add(new HttpKafkaWithFetchFilterConfig(newKey, newHeaders));
+                newFilters.add(HttpKafkaWithFetchFilterConfig.builder()
+                    .key(newKey)
+                    .headers(newHeaders)
+                    .build());
             }
         }
 

runtime/binding-http-kafka/src/main/java/io/aklivity/zilla/runtime/binding/http/kafka/internal/config/HttpKafkaWithResolver.java

@@ -31,6 +31,8 @@
 import io.aklivity.zilla.runtime.binding.http.kafka.config.HttpKafkaOptionsConfig;
 import io.aklivity.zilla.runtime.binding.http.kafka.config.HttpKafkaWithConfig;
 import io.aklivity.zilla.runtime.binding.http.kafka.config.HttpKafkaWithFetchConfig;
+import io.aklivity.zilla.runtime.binding.http.kafka.config.HttpKafkaWithFetchFilterConfig;
+import io.aklivity.zilla.runtime.binding.http.kafka.config.HttpKafkaWithFetchFilterHeaderConfig;
 import io.aklivity.zilla.runtime.binding.http.kafka.config.HttpKafkaWithFetchMergeConfig;
 import io.aklivity.zilla.runtime.binding.http.kafka.config.HttpKafkaWithProduceAsyncHeaderConfig;
 import io.aklivity.zilla.runtime.binding.http.kafka.config.HttpKafkaWithProduceConfig;

runtime/binding-http-kafka/src/test/java/io/aklivity/zilla/runtime/binding/http/kafka/internal/config/HttpKafkaWithConfigAdapterTest.java

@@ -37,6 +37,7 @@
 
 import io.aklivity.zilla.runtime.binding.http.kafka.config.HttpKafkaWithConfig;
 import io.aklivity.zilla.runtime.binding.http.kafka.config.HttpKafkaWithFetchConfig;
+import io.aklivity.zilla.runtime.binding.http.kafka.config.HttpKafkaWithFetchFilterConfig;
 import io.aklivity.zilla.runtime.binding.http.kafka.config.HttpKafkaWithFetchMergeConfig;
 import io.aklivity.zilla.runtime.binding.http.kafka.config.HttpKafkaWithProduceAsyncHeaderConfig;
 import io.aklivity.zilla.runtime.binding.http.kafka.config.HttpKafkaWithProduceConfig;
@@ -129,11 +130,10 @@ public void shouldWriteWithFetchTopicAndFilters()
         HttpKafkaWithConfig with = HttpKafkaWithConfig.builder()
             .fetch(HttpKafkaWithFetchConfig.builder()
                 .topic("test")
-                .filters(singletonList(new HttpKafkaWithFetchFilterConfig(
-                        "fixed-key",
-                        singletonList(new HttpKafkaWithFetchFilterHeaderConfig(
-                            "tag",
-                            "fixed-tag")))))
+                .filters(singletonList(HttpKafkaWithFetchFilterConfig.builder()
+                    .key("fixed-key")
+                    .header("tag", "fixed-tag")
+                    .build()))
                 .merged(null)
                 .build())
             .build();