Support identity placeholder in MQTT binding topics

aklivity · Dec 27, 2024 · 3bd97af · 3bd97af
1 parent 3e2a862
commit 3bd97af
Show file tree

Hide file tree

Showing 2 changed files with 26 additions and 6 deletions.
diff --git a/...c/main/java/io/aklivity/zilla/runtime/binding/mqtt/internal/config/MqttBindingConfig.java b/...c/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(topic, authorization))
             .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(topic, authorization))
             .findFirst()
             .orElse(null);
     }

diff --git a/...src/main/java/io/aklivity/zilla/runtime/binding/mqtt/internal/config/MqttRouteConfig.java b/...src/main/java/io/aklivity/zilla/runtime/binding/mqtt/internal/config/MqttRouteConfig.java
@@ -23,6 +23,7 @@
 
 import io.aklivity.zilla.runtime.binding.mqtt.config.MqttConditionConfig;
 import io.aklivity.zilla.runtime.binding.mqtt.config.MqttWithConfig;
+import io.aklivity.zilla.runtime.engine.config.GuardedConfig;
 import io.aklivity.zilla.runtime.engine.config.RouteConfig;
 
 public final class MqttRouteConfig
@@ -32,6 +33,7 @@ public final class MqttRouteConfig
     private final List<MqttConditionMatcher> when;
     private final MqttWithConfig with;
     private final LongPredicate authorized;
+    private final List<GuardedConfig> guarded;
 
     public MqttRouteConfig(
         RouteConfig route)
@@ -43,6 +45,7 @@ public MqttRouteConfig(
             .collect(toList());
         this.with = (MqttWithConfig) route.with;
         this.authorized = route.authorized;
+        this.guarded = route.guarded;
     }
 
     public long compositeId()
@@ -63,14 +66,31 @@ boolean matchesSession(
     }
 
     boolean matchesSubscribe(
-        String topic)
+        String topic,
+        long authorization)
     {
-        return when.isEmpty() || when.stream().anyMatch(m -> m.matchesSubscribe(topic));
+        String actualTopic = replaceTopicPlaceholders(topic, authorization);
+        return when.isEmpty() || when.stream().anyMatch(m -> m.matchesSubscribe(actualTopic));
     }
 
     boolean matchesPublish(
-        String topic)
+        String topic,
+        long authorization)
+    {
+        String actualTopic = replaceTopicPlaceholders(topic, authorization);
+        return when.isEmpty() || when.stream().anyMatch(m -> m.matchesPublish(actualTopic));
+    }
+
+    private String replaceTopicPlaceholders(String topic, long authorization)
     {
-        return when.isEmpty() || when.stream().anyMatch(m -> m.matchesPublish(topic));
+        for (GuardedConfig g : guarded)
+        {
+            String identity = g.identity.apply(authorization);
+            if (identity != null)
+            {
+                topic = topic.replace(String.format("${guarded[%s].identity}", g.name), identity);
+            }
+        }
+        return topic;
     }
 }