MQTT topics with path parameters required to match guarded identity

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

@@ -17,12 +17,15 @@
 
 import static java.util.function.Function.identity;
 
+import java.util.List;
 import java.util.function.Function;
 
 public class MqttPublishConfig
 {
     public final String topic;
 
+    public final List<MqttTopicParamConfig> params;
+
     public static MqttPublishConfigBuilder<MqttPublishConfig> builder()
     {
         return new MqttPublishConfigBuilder<>(identity());
@@ -35,8 +38,10 @@ public static <T> MqttPublishConfigBuilder<T> builder(
     }
 
     MqttPublishConfig(
-        String topic)
+        String topic,
+        List<MqttTopicParamConfig> params)
     {
         this.topic = topic;
+        this.params = params;
     }
 }

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

@@ -15,6 +15,8 @@
  */
 package io.aklivity.zilla.runtime.binding.mqtt.config;
 
+import java.util.ArrayList;
+import java.util.List;
 import java.util.function.Function;
 
 import io.aklivity.zilla.runtime.engine.config.ConfigBuilder;
@@ -25,6 +27,8 @@ public class MqttPublishConfigBuilder<T> extends ConfigBuilder<T, MqttPublishCon
 
     private String topic;
 
+    private List<MqttTopicParamConfig> params;
+
     MqttPublishConfigBuilder(
         Function<MqttPublishConfig, T> mapper)
     {
@@ -45,9 +49,25 @@ public MqttPublishConfigBuilder<T> topic(
         return this;
     }
 
+    public MqttPublishConfigBuilder<T> param(
+        MqttTopicParamConfig param)
+    {
+        if (this.params == null)
+        {
+            this.params = new ArrayList<>();
+        }
+        this.params.add(param);
+        return this;
+    }
+
+    public MqttTopicParamConfigBuilder<MqttPublishConfigBuilder<T>> param()
+    {
+        return new MqttTopicParamConfigBuilder<>(this::param);
+    }
+
     @Override
     public T build()
     {
-        return mapper.apply(new MqttPublishConfig(topic));
+        return mapper.apply(new MqttPublishConfig(topic, params));
     }
 }

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

@@ -17,12 +17,15 @@
 
 import static java.util.function.Function.identity;
 
+import java.util.List;
 import java.util.function.Function;
 
 public class MqttSubscribeConfig
 {
     public final String topic;
 
+    public final List<MqttTopicParamConfig> params;
+
     public static MqttSubscribeConfigBuilder<MqttSubscribeConfig> builder()
     {
         return new MqttSubscribeConfigBuilder<>(identity());
@@ -35,8 +38,10 @@ public static <T> MqttSubscribeConfigBuilder<T> builder(
     }
 
     MqttSubscribeConfig(
-        String topic)
+        String topic,
+        List<MqttTopicParamConfig> params)
     {
         this.topic = topic;
+        this.params = params;
     }
 }

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

@@ -15,6 +15,8 @@
  */
 package io.aklivity.zilla.runtime.binding.mqtt.config;
 
+import java.util.ArrayList;
+import java.util.List;
 import java.util.function.Function;
 
 import io.aklivity.zilla.runtime.engine.config.ConfigBuilder;
@@ -25,6 +27,8 @@ public class MqttSubscribeConfigBuilder<T> extends ConfigBuilder<T, MqttSubscrib
 
     private String topic;
 
+    private List<MqttTopicParamConfig> params;
+
     MqttSubscribeConfigBuilder(
         Function<MqttSubscribeConfig, T> mapper)
     {
@@ -45,9 +49,25 @@ public MqttSubscribeConfigBuilder<T> topic(
         return this;
     }
 
+    public MqttSubscribeConfigBuilder<T> param(
+        MqttTopicParamConfig param)
+    {
+        if (this.params == null)
+        {
+            this.params = new ArrayList<>();
+        }
+        this.params.add(param);
+        return this;
+    }
+
+    public MqttTopicParamConfigBuilder<MqttSubscribeConfigBuilder<T>> param()
+    {
+        return new MqttTopicParamConfigBuilder<>(this::param);
+    }
+
     @Override
     public T build()
     {
-        return mapper.apply(new MqttSubscribeConfig(topic));
+        return mapper.apply(new MqttSubscribeConfig(topic, params));
     }
 }

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

@@ -0,0 +1,46 @@
+/*
+ * Copyright 2021-2024 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.mqtt.config;
+
+import static java.util.function.Function.identity;
+
+import java.util.function.Function;
+
+public class MqttTopicParamConfig
+{
+    public final String name;
+
+    public final String value;
+
+    public static MqttTopicParamConfigBuilder<MqttTopicParamConfig> builder()
+    {
+        return new MqttTopicParamConfigBuilder<>(identity());
+    }
+
+    public static <T> MqttTopicParamConfigBuilder<T> builder(
+        Function<MqttTopicParamConfig, T> mapper)
+    {
+        return new MqttTopicParamConfigBuilder<>(mapper);
+    }
+
+    MqttTopicParamConfig(
+        String name,
+        String value)
+    {
+        this.name = name;
+        this.value = value;
+    }
+}

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

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

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

@@ -118,7 +118,7 @@ public MqttRouteConfig resolveSubscribe(
         String topic)
     {
         return routes.stream()
-            .filter(r -> r.authorized(authorization) && r.matchesSubscribe(topic))
+            .filter(r -> r.authorized(authorization) && r.matchesSubscribe(authorization, topic))
             .findFirst()
             .orElse(null);
     }
@@ -128,7 +128,7 @@ public MqttRouteConfig resolvePublish(
         String topic)
     {
         return routes.stream()
-            .filter(r -> r.authorized(authorization) && r.matchesPublish(topic))
+            .filter(r -> r.authorized(authorization) && r.matchesPublish(authorization, topic))
             .findFirst()
             .orElse(null);
     }

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

@@ -24,6 +24,8 @@
 
 import io.aklivity.zilla.runtime.binding.mqtt.config.MqttConditionConfig;
 import io.aklivity.zilla.runtime.binding.mqtt.config.MqttConditionConfigBuilder;
+import io.aklivity.zilla.runtime.binding.mqtt.config.MqttPublishConfigBuilder;
+import io.aklivity.zilla.runtime.binding.mqtt.config.MqttSubscribeConfigBuilder;
 import io.aklivity.zilla.runtime.binding.mqtt.internal.MqttBinding;
 import io.aklivity.zilla.runtime.engine.config.ConditionConfig;
 import io.aklivity.zilla.runtime.engine.config.ConditionConfigAdapterSpi;
@@ -35,6 +37,7 @@ public final class MqttConditionConfigAdapter implements ConditionConfigAdapterS
     private static final String PUBLISH_NAME = "publish";
     private static final String CLIENT_ID_NAME = "client-id";
     private static final String TOPIC_NAME = "topic";
+    private static final String PARAMS_NAME = "params";
 
     private static final String CLIENT_ID_DEFAULT = "*";
 
@@ -72,10 +75,16 @@ public JsonObject adaptToJson(
         {
             JsonArrayBuilder subscribes = Json.createArrayBuilder();
 
-            mqttCondition.subscribes.forEach(s ->
+            mqttCondition.subscribes.forEach(sub ->
             {
                 JsonObjectBuilder subscribeJson = Json.createObjectBuilder();
-                subscribeJson.add(TOPIC_NAME, s.topic);
+                subscribeJson.add(TOPIC_NAME, sub.topic);
+                if (sub.params != null)
+                {
+                    JsonObjectBuilder params = Json.createObjectBuilder();
+                    sub.params.forEach(p -> params.add(p.name, p.value));
+                    subscribeJson.add(PARAMS_NAME, params);
+                }
                 subscribes.add(subscribeJson);
             });
             object.add(SUBSCRIBE_NAME, subscribes);
@@ -85,10 +94,16 @@ public JsonObject adaptToJson(
         {
             JsonArrayBuilder publishes = Json.createArrayBuilder();
 
-            mqttCondition.publishes.forEach(p ->
+            mqttCondition.publishes.forEach(pub ->
             {
                 JsonObjectBuilder publishJson = Json.createObjectBuilder();
-                publishJson.add(TOPIC_NAME, p.topic);
+                publishJson.add(TOPIC_NAME, pub.topic);
+                if (pub.params != null)
+                {
+                    JsonObjectBuilder params = Json.createObjectBuilder();
+                    pub.params.forEach(p -> params.add(p.name, p.value));
+                    publishJson.add(PARAMS_NAME, params);
+                }
                 publishes.add(publishJson);
             });
             object.add(PUBLISH_NAME, publishes);
@@ -121,11 +136,24 @@ public ConditionConfig adaptFromJson(
             JsonArray subscribesJson = object.getJsonArray(SUBSCRIBE_NAME);
             subscribesJson.forEach(s ->
             {
-                String topic = s.asJsonObject().getString(TOPIC_NAME);
+                JsonObject subscribeJson = s.asJsonObject();
 
-                mqttConfig.subscribe()
-                    .topic(topic)
-                    .build();
+                MqttSubscribeConfigBuilder<?> subscribe = mqttConfig
+                    .subscribe()
+                    .topic(subscribeJson.getString(TOPIC_NAME));
+
+                if (subscribeJson.containsKey(PARAMS_NAME))
+                {
+                    JsonObject paramsJson = subscribeJson.getJsonObject(PARAMS_NAME);
+
+                    paramsJson.keySet().forEach(n ->
+                        subscribe.param()
+                            .name(n)
+                            .value(paramsJson.getString(n))
+                            .build());
+                }
+
+                subscribe.build();
             });
         }
 
@@ -134,11 +162,24 @@ public ConditionConfig adaptFromJson(
             JsonArray publishesJson = object.getJsonArray(PUBLISH_NAME);
             publishesJson.forEach(p ->
             {
-                String topic = p.asJsonObject().getString(TOPIC_NAME);
+                JsonObject publishJson = p.asJsonObject();
 
-                mqttConfig.publish()
-                    .topic(topic)
-                    .build();
+                MqttPublishConfigBuilder<?> publish = mqttConfig
+                    .publish()
+                    .topic(publishJson.getString(TOPIC_NAME));
+
+                if (publishJson.containsKey(PARAMS_NAME))
+                {
+                    JsonObject paramsJson = publishJson.getJsonObject(PARAMS_NAME);
+
+                    paramsJson.keySet().forEach(n ->
+                        publish.param()
+                            .name(n)
+                            .value(paramsJson.getString(n))
+                            .build());
+                }
+
+                publish.build();
             });
         }