support dynamic guarded route (#1478)

Author: ankitk-me

incubator/binding-amqp/src/main/java/io/aklivity/zilla/runtime/binding/amqp/internal/config/AmqpRouteConfig.java

@@ -15,21 +15,23 @@
  */
 package io.aklivity.zilla.runtime.binding.amqp.internal.config;
 
+import static java.util.function.UnaryOperator.identity;
 import static java.util.stream.Collectors.toList;
 
 import java.util.List;
-import java.util.function.LongPredicate;
+import java.util.function.UnaryOperator;
 
 import io.aklivity.zilla.runtime.binding.amqp.config.AmqpConditionConfig;
 import io.aklivity.zilla.runtime.binding.amqp.internal.types.AmqpCapabilities;
 import io.aklivity.zilla.runtime.engine.config.RouteConfig;
+import io.aklivity.zilla.runtime.engine.util.function.LongObjectPredicate;
 
 public final class AmqpRouteConfig
 {
     public final long id;
 
     private final List<AmqpConditionMatcher> when;
-    private final LongPredicate authorized;
+    private final LongObjectPredicate<UnaryOperator<String>> authorized;
 
     public AmqpRouteConfig(
         RouteConfig route)
@@ -45,7 +47,7 @@ public AmqpRouteConfig(
     boolean authorized(
         long authorization)
     {
-        return authorized.test(authorization);
+        return authorized.test(authorization, identity());
     }
 
     boolean matches(

incubator/binding-pgsql-kafka/src/main/java/io/aklivity/zilla/runtime/binding/pgsql/kafka/internal/config/PgsqlKafkaRouteConfig.java

@@ -14,16 +14,19 @@
  */
 package io.aklivity.zilla.runtime.binding.pgsql.kafka.internal.config;
 
-import java.util.function.LongPredicate;
+import static java.util.function.UnaryOperator.identity;
+
+import java.util.function.UnaryOperator;
 
 import io.aklivity.zilla.runtime.engine.config.RouteConfig;
+import io.aklivity.zilla.runtime.engine.util.function.LongObjectPredicate;
 
 public final class PgsqlKafkaRouteConfig
 {
     public final long id;
     public final int order;
 
-    private final LongPredicate authorized;
+    private final LongObjectPredicate<UnaryOperator<String>> authorized;
 
     public PgsqlKafkaRouteConfig(
         RouteConfig route)
@@ -36,6 +39,6 @@ public PgsqlKafkaRouteConfig(
     boolean authorized(
         long authorization)
     {
-        return authorized.test(authorization);
+        return authorized.test(authorization, identity());
     }
 }

incubator/binding-pgsql/src/main/java/io/aklivity/zilla/runtime/binding/pgsql/internal/config/PgsqlRouteConfig.java

@@ -14,16 +14,19 @@
  */
 package io.aklivity.zilla.runtime.binding.pgsql.internal.config;
 
-import java.util.function.LongPredicate;
+import static java.util.function.UnaryOperator.identity;
+
+import java.util.function.UnaryOperator;
 
 import io.aklivity.zilla.runtime.engine.config.RouteConfig;
+import io.aklivity.zilla.runtime.engine.util.function.LongObjectPredicate;
 
 public final class PgsqlRouteConfig
 {
     public final long id;
     public final int order;
 
-    private final LongPredicate authorized;
+    private final LongObjectPredicate<UnaryOperator<String>> authorized;
 
     public PgsqlRouteConfig(
         RouteConfig route)
@@ -36,6 +39,6 @@ public PgsqlRouteConfig(
     boolean authorized(
         long authorization)
     {
-        return authorized.test(authorization);
+        return authorized.test(authorization, identity());
     }
 }

incubator/binding-risingwave/src/main/java/io/aklivity/zilla/runtime/binding/risingwave/internal/config/RisingwaveRouteConfig.java

@@ -14,22 +14,24 @@
  */
 package io.aklivity.zilla.runtime.binding.risingwave.internal.config;
 
+import static java.util.function.UnaryOperator.identity;
 import static java.util.stream.Collectors.toList;
 
 import java.util.List;
-import java.util.function.LongPredicate;
+import java.util.function.UnaryOperator;
 
 import org.agrona.DirectBuffer;
 
 import io.aklivity.zilla.runtime.binding.risingwave.config.RisingwaveConditionConfig;
 import io.aklivity.zilla.runtime.engine.config.RouteConfig;
+import io.aklivity.zilla.runtime.engine.util.function.LongObjectPredicate;
 
 public final class RisingwaveRouteConfig
 {
     public final long id;
 
     private final List<RisingwaveConditionMatcher> when;
-    private final LongPredicate authorized;
+    private final LongObjectPredicate<UnaryOperator<String>> authorized;
 
     public RisingwaveRouteConfig(
         RouteConfig route)
@@ -45,7 +47,7 @@ public RisingwaveRouteConfig(
     boolean authorized(
         long authorization)
     {
-        return authorized.test(authorization);
+        return authorized.test(authorization, identity());
     }
 
     boolean matches(

runtime/binding-asyncapi/src/main/java/io/aklivity/zilla/runtime/binding/asyncapi/internal/config/AsyncapiRouteConfig.java

@@ -14,20 +14,22 @@
  */
 package io.aklivity.zilla.runtime.binding.asyncapi.internal.config;
 
+import static java.util.function.UnaryOperator.identity;
 import static java.util.stream.Collectors.toList;
 
 import java.util.List;
-import java.util.function.LongPredicate;
+import java.util.function.UnaryOperator;
 
 import io.aklivity.zilla.runtime.engine.config.RouteConfig;
+import io.aklivity.zilla.runtime.engine.util.function.LongObjectPredicate;
 
 public final class AsyncapiRouteConfig
 {
     public final long id;
     public final List<AsyncapiConditionConfig> when;
     public final AsyncapiWithConfig with;
 
-    private final LongPredicate authorized;
+    private final LongObjectPredicate<UnaryOperator<String>> authorized;
 
     public AsyncapiRouteConfig(
         RouteConfig route)
@@ -43,7 +45,7 @@ public AsyncapiRouteConfig(
     boolean authorized(
         long authorization)
     {
-        return authorized.test(authorization);
+        return authorized.test(authorization, identity());
     }
 
     boolean matches(

runtime/binding-fan/src/main/java/io/aklivity/zilla/runtime/binding/fan/internal/stream/FanServerFactory.java

@@ -18,6 +18,7 @@
 import static io.aklivity.zilla.runtime.engine.budget.BudgetCreditor.NO_BUDGET_ID;
 import static io.aklivity.zilla.runtime.engine.budget.BudgetCreditor.NO_CREDITOR_INDEX;
 import static io.aklivity.zilla.runtime.engine.budget.BudgetDebitor.NO_DEBITOR_INDEX;
+import static java.util.function.UnaryOperator.identity;
 
 import java.util.List;
 import java.util.concurrent.CopyOnWriteArrayList;
@@ -132,7 +133,7 @@ public MessageConsumer newStream(
             final long replyId = supplyReplyId.applyAsLong(initialId);
 
             final RouteConfig route = binding.routes.stream()
-                    .filter(r -> r.authorized.test(authorization))
+                    .filter(r -> r.authorized.test(authorization, identity()))
                     .findFirst()
                     .orElse(null);
 

runtime/binding-grpc-kafka/src/main/java/io/aklivity/zilla/runtime/binding/grpc/kafka/internal/config/GrpcKafkaRouteConfig.java

@@ -14,13 +14,14 @@
  */
 package io.aklivity.zilla.runtime.binding.grpc.kafka.internal.config;
 
+import static java.util.function.UnaryOperator.identity;
 import static java.util.stream.Collectors.toList;
 
 import java.util.List;
 import java.util.Map;
 import java.util.Optional;
 import java.util.function.LongFunction;
-import java.util.function.LongPredicate;
+import java.util.function.UnaryOperator;
 import java.util.regex.MatchResult;
 import java.util.stream.Collectors;
 
@@ -31,6 +32,7 @@
 import io.aklivity.zilla.runtime.binding.grpc.kafka.internal.types.stream.GrpcMetadataFW;
 import io.aklivity.zilla.runtime.engine.config.RouteConfig;
 import io.aklivity.zilla.runtime.engine.util.function.LongObjectBiFunction;
+import io.aklivity.zilla.runtime.engine.util.function.LongObjectPredicate;
 
 
 public final class GrpcKafkaRouteConfig
@@ -39,7 +41,7 @@ public final class GrpcKafkaRouteConfig
 
     private final List<GrpcKafkaConditionMatcher> when;
     public final GrpcKafkaWithResolver with;
-    private final LongPredicate authorized;
+    private final LongObjectPredicate<UnaryOperator<String>> authorized;
 
     public GrpcKafkaRouteConfig(
         GrpcKafkaOptionsConfig options,
@@ -72,7 +74,7 @@ public GrpcKafkaRouteConfig(
     boolean authorized(
         long authorization)
     {
-        return authorized.test(authorization);
+        return authorized.test(authorization, identity());
     }
 
     boolean matches(

runtime/binding-grpc/src/main/java/io/aklivity/zilla/runtime/binding/grpc/internal/config/GrpcRouteConfig.java

@@ -14,22 +14,24 @@
  */
 package io.aklivity.zilla.runtime.binding.grpc.internal.config;
 
+import static java.util.function.UnaryOperator.identity;
 import static java.util.stream.Collectors.toList;
 
 import java.util.List;
-import java.util.function.LongPredicate;
+import java.util.function.UnaryOperator;
 
 import io.aklivity.zilla.runtime.binding.grpc.config.GrpcConditionConfig;
 import io.aklivity.zilla.runtime.binding.grpc.internal.types.Array32FW;
 import io.aklivity.zilla.runtime.binding.grpc.internal.types.stream.GrpcMetadataFW;
 import io.aklivity.zilla.runtime.engine.config.RouteConfig;
+import io.aklivity.zilla.runtime.engine.util.function.LongObjectPredicate;
 
 public final class GrpcRouteConfig
 {
     public final long id;
 
     private final List<GrpcConditionMatcher> when;
-    private final LongPredicate authorized;
+    private final LongObjectPredicate<UnaryOperator<String>> authorized;
 
     public GrpcRouteConfig(
         RouteConfig route)
@@ -45,7 +47,7 @@ public GrpcRouteConfig(
     boolean authorized(
         long authorization)
     {
-        return authorized.test(authorization);
+        return authorized.test(authorization, identity());
     }
 
     boolean matches(

runtime/binding-http-filesystem/src/main/java/io/aklivity/zilla/runtime/binding/http/filesystem/internal/config/HttpFileSystemRouteConfig.java

@@ -14,23 +14,25 @@
  */
 package io.aklivity.zilla.runtime.binding.http.filesystem.internal.config;
 
+import static java.util.function.UnaryOperator.identity;
 import static java.util.stream.Collectors.toList;
 
 import java.util.List;
 import java.util.Optional;
 import java.util.function.Consumer;
-import java.util.function.LongPredicate;
+import java.util.function.UnaryOperator;
 
 import io.aklivity.zilla.runtime.binding.http.filesystem.config.HttpFileSystemConditionConfig;
 import io.aklivity.zilla.runtime.engine.config.RouteConfig;
+import io.aklivity.zilla.runtime.engine.util.function.LongObjectPredicate;
 
 public final class HttpFileSystemRouteConfig
 {
     public final long id;
     public final Optional<HttpFileSystemWithResolver> with;
 
     private final List<HttpFileSystemConditionMatcher> when;
-    private final LongPredicate authorized;
+    private final LongObjectPredicate<UnaryOperator<String>> authorized;
 
     public HttpFileSystemRouteConfig(
         RouteConfig route)
@@ -51,7 +53,7 @@ public HttpFileSystemRouteConfig(
     boolean authorized(
         long authorization)
     {
-        return authorized.test(authorization);
+        return authorized.test(authorization, identity());
     }
 
     boolean matches(

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

@@ -66,7 +66,7 @@ public HttpKafkaRouteConfig resolve(
         CharSequence path = helper.path;
 
         return routes.stream()
-            .filter(r -> r.authorized(authorization) && r.matches(method, path))
+            .filter(r -> r.authorized(authorization, method, path) && r.matches(method, path))
             .findFirst()
             .orElse(null);
     }