Merge upstream changes from develop to feature.

hiddenalpha · Jun 6, 2024 · 8397965 · 8397965
2 parents c9a02c8 + cdceaa0
commit 8397965
Show file tree

Hide file tree

Showing 8 changed files with 65 additions and 407 deletions.
diff --git a/gateleen-playground/src/main/java/org/swisspush/gateleen/playground/Server.java b/gateleen-playground/src/main/java/org/swisspush/gateleen/playground/Server.java
@@ -1,6 +1,9 @@
 package org.swisspush.gateleen.playground;
 
-import io.vertx.core.*;
+import io.vertx.core.AbstractVerticle;
+import io.vertx.core.Future;
+import io.vertx.core.Handler;
+import io.vertx.core.Vertx;
 import io.vertx.core.http.HttpClient;
 import io.vertx.core.http.HttpClientOptions;
 import io.vertx.core.http.HttpServer;
@@ -55,7 +58,6 @@
 import org.swisspush.gateleen.monitoring.MonitoringHandler;
 import org.swisspush.gateleen.monitoring.ResetMetricsController;
 import org.swisspush.gateleen.qos.QoSHandler;
-import org.swisspush.gateleen.queue.queuing.QueueBrowser;
 import org.swisspush.gateleen.queue.queuing.QueueClient;
 import org.swisspush.gateleen.queue.queuing.QueueProcessor;
 import org.swisspush.gateleen.queue.queuing.circuitbreaker.QueueCircuitBreaker;
@@ -332,8 +334,6 @@ public void start() {
                         queueCircuitBreakerConfigurationResourceManager, requestHandler, circuitBreakerPort);
 
                 new QueueProcessor(vertx, selfClient, monitoringHandler, queueCircuitBreaker);
-                final QueueBrowser queueBrowser = new QueueBrowser(vertx, SERVER_ROOT + "/queuing", Address.redisquesAddress(),
-                        monitoringHandler);
 
                 LogController logController = new LogController();
                 logController.registerLogConfiguratorMBean(JMX_DOMAIN);
@@ -365,7 +365,7 @@ public void start() {
                         .delegateHandler(delegateHandler)
                         .customHttpResponseHandler(customHttpResponseHandler)
                         .contentTypeConstraintHandler(contentTypeConstraintHandler)
-                        .build(vertx, redisProvider, Server.class, router, monitoringHandler, queueBrowser);
+                        .build(vertx, redisProvider, Server.class, router, monitoringHandler);
                 Handler<RoutingContext> routingContextHandlerrNew = runConfig.buildRoutingContextHandler();
                 selfClient.setRoutingContexttHandler(routingContextHandlerrNew);
 

diff --git a/gateleen-queue/src/main/java/org/swisspush/gateleen/queue/queuing/QueueBrowser.java b/gateleen-queue/src/main/java/org/swisspush/gateleen/queue/queuing/QueueBrowser.java
diff --git a/gateleen-runconfig/src/main/java/org/swisspush/gateleen/runconfig/RunConfig.java b/gateleen-runconfig/src/main/java/org/swisspush/gateleen/runconfig/RunConfig.java
@@ -35,7 +35,6 @@
 import org.swisspush.gateleen.monitoring.MonitoringHandler;
 import org.swisspush.gateleen.packing.PackingHandler;
 import org.swisspush.gateleen.qos.QoSHandler;
-import org.swisspush.gateleen.queue.queuing.QueueBrowser;
 import org.swisspush.gateleen.queue.queuing.QueuingHandler;
 import org.swisspush.gateleen.queue.queuing.circuitbreaker.configuration.QueueCircuitBreakerConfigurationResourceManager;
 import org.swisspush.gateleen.queue.queuing.splitter.QueueSplitter;
@@ -101,7 +100,6 @@ public class RunConfig {
     private final ExpansionHandler expansionHandler;
     private final DeltaHandler deltaHandler;
     private final MonitoringHandler monitoringHandler;
-    private final QueueBrowser queueBrowser;
     private final Authorizer authorizer;
     private final CopyResourceHandler copyResourceHandler;
     private final QoSHandler qosHandler;
@@ -113,7 +111,7 @@ public class RunConfig {
     private final CustomHttpResponseHandler customHttpResponseHandler;
 
     public RunConfig(Vertx vertx, RedisProvider redisProvider, Class verticleClass, Router router, MonitoringHandler monitoringHandler,
-                     QueueBrowser queueBrowser, CORSHandler corsHandler, SchedulerResourceManager schedulerResourceManager,
+                     CORSHandler corsHandler, SchedulerResourceManager schedulerResourceManager,
                      ValidationResourceManager validationResourceManager, LoggingResourceManager loggingResourceManager,
                      ConfigurationResourceManager configurationResourceManager,
                      QueueCircuitBreakerConfigurationResourceManager queueCircuitBreakerConfigurationResourceManager,
@@ -129,7 +127,6 @@ public RunConfig(Vertx vertx, RedisProvider redisProvider, Class verticleClass,
         this.verticleClass = verticleClass;
         this.router = router;
         this.monitoringHandler = monitoringHandler;
-        this.queueBrowser = queueBrowser;
         this.corsHandler = corsHandler;
         this.schedulerResourceManager = schedulerResourceManager;
         this.validationResourceManager = validationResourceManager;
@@ -164,7 +161,6 @@ private RunConfig(RunConfigBuilder builder) {
                 builder.verticleClass,
                 builder.router,
                 builder.monitoringHandler,
-                builder.queueBrowser,
                 builder.corsHandler,
                 builder.schedulerResourceManager,
                 builder.validationResourceManager,
@@ -222,7 +218,6 @@ public static class RunConfigBuilder {
         private Class verticleClass;
         private Router router;
         private MonitoringHandler monitoringHandler;
-        private QueueBrowser queueBrowser;
         private CORSHandler corsHandler;
         private SchedulerResourceManager schedulerResourceManager;
         private ValidationResourceManager validationResourceManager;
@@ -377,13 +372,12 @@ public RunConfigBuilder cacheHandler(CacheHandler cacheHandler) {
             return this;
         }
 
-        public RunConfig build(Vertx vertx, RedisProvider redisProvider, Class verticleClass, Router router, MonitoringHandler monitoringHandler, QueueBrowser queueBrowser) {
+        public RunConfig build(Vertx vertx, RedisProvider redisProvider, Class verticleClass, Router router, MonitoringHandler monitoringHandler) {
             this.vertx = vertx;
             this.redisProvider = redisProvider;
             this.verticleClass = verticleClass;
             this.router = router;
             this.monitoringHandler = monitoringHandler;
-            this.queueBrowser = queueBrowser;
             return new RunConfig(this);
         }
     }
@@ -633,10 +627,6 @@ private void handleRequest(final RoutingContext ctx) {
                         if (copyResourceHandler != null && copyResourceHandler.handle(request)) {
                             return;
                         }
-                        if (request.path().startsWith(SERVER_ROOT + "/queuing/")) {
-                            queueBrowser.handle(request);
-                            return;
-                        }
                         if (hookHandler != null && hookHandler.handle(ctx)) {
                             return;
                         }

diff --git a/gateleen-test/src/test/java/org/swisspush/gateleen/AbstractTest.java b/gateleen-test/src/test/java/org/swisspush/gateleen/AbstractTest.java
@@ -57,7 +57,6 @@
 import org.swisspush.gateleen.monitoring.MonitoringHandler;
 import org.swisspush.gateleen.monitoring.ResetMetricsController;
 import org.swisspush.gateleen.qos.QoSHandler;
-import org.swisspush.gateleen.queue.queuing.QueueBrowser;
 import org.swisspush.gateleen.queue.queuing.QueueClient;
 import org.swisspush.gateleen.queue.queuing.QueueProcessor;
 import org.swisspush.gateleen.queue.queuing.circuitbreaker.QueueCircuitBreaker;
@@ -105,6 +104,7 @@ public abstract class AbstractTest {
     protected static final int REDIS_PORT = 6379;
     protected static final int STORAGE_PORT = 8989;
     private static final int CIRCUIT_BREAKER_REST_API_PORT = 7014;
+    protected static final int REDISQUES_API_PORT = 7015;
 
     /**
      * Basis configuration for RestAssured
@@ -219,8 +219,6 @@ public static void setupBeforeClass(TestContext context) {
                         CIRCUIT_BREAKER_REST_API_PORT);
 
                 new QueueProcessor(vertx, selfClient, monitoringHandler, queueCircuitBreaker);
-                final QueueBrowser queueBrowser = new QueueBrowser(vertx, SERVER_ROOT + "/queuing",
-                        Address.redisquesAddress(), monitoringHandler);
 
                 new CustomRedisMonitor(vertx, redisProvider, "main", "rest-storage", 10).start();
                 Router router = Router.builder()
@@ -266,7 +264,7 @@ public static void setupBeforeClass(TestContext context) {
                                 .delegateHandler(delegateHandler)
                                 .mergeHandler(mergeHandler)
                                 .customHttpResponseHandler(customHttpResponseHandler)
-                                .build(vertx, redisProvider, AbstractTest.class, router, monitoringHandler, queueBrowser);
+                                .build(vertx, redisProvider, AbstractTest.class, router, monitoringHandler);
                 Handler<RoutingContext> routingContextHandlerrNew = runConfig.buildRoutingContextHandler();
                 selfClient.setRoutingContexttHandler(routingContextHandlerrNew);
                 mainServer = vertx.createHttpServer();

diff --git a/gateleen-test/src/test/java/org/swisspush/gateleen/TestUtils.java b/gateleen-test/src/test/java/org/swisspush/gateleen/TestUtils.java
@@ -8,10 +8,8 @@
 import org.swisspush.gateleen.hook.HookHandler;
 import org.swisspush.gateleen.hook.HookTriggerType;
 
-import java.util.Arrays;
-import java.util.Map;
+import java.util.*;
 import java.util.Map.Entry;
-import java.util.Set;
 import java.util.regex.Pattern;
 
 import static org.awaitility.Awaitility.await;
@@ -114,6 +112,23 @@ public static void waitSomeTime(int seconds) {
         }
     }
 
+    /**
+     * Adds the routing rule for queuing to the given routing rules.
+     *
+     * @param rules current rules.
+     */
+    public static JsonObject addRoutingRuleQueuing(JsonObject rules) {
+
+        JsonObject queuing = createRoutingRule(ImmutableMap.of(
+                "description",
+                "vertx-redisques API",
+                "url",
+                "http://localhost:" + AbstractTest.REDISQUES_API_PORT + "/queuing/$1"));
+
+        rules = addRoutingRule(rules, AbstractTest.SERVER_ROOT + "/queuing/(.*)", queuing);
+        return rules;
+    }
+
     /**
      * Adds the routing rule for cleanup to the given routing rules.
      * 
@@ -348,9 +363,29 @@ public static void registerListener(final String requestUrl, final String target
     public static void registerListener(final String requestUrl, final String target, String[] methods, String filter,
                                         Integer queueExpireTime, Map<String, String> staticHeaders, HookTriggerType type,
                                         String headersFilter) {
+        registerListener(requestUrl, target, methods, filter, queueExpireTime, staticHeaders, type, headersFilter, null);
+    }
+
+    /**
+     * Registers a listener with a filter, static headers, a queue header and a event trigger.
+     *
+     * @param requestUrl
+     * @param target
+     * @param methods
+     * @param filter
+     * @param queueExpireTime
+     * @param staticHeaders
+     * @param queueHeader
+     */
+    public static void registerListener(final String requestUrl, final String target, String[] methods, String filter,
+                                        Integer queueExpireTime, Map<String, String> staticHeaders, HookTriggerType type,
+                                        String headersFilter, String queueHeader) {
         JsonObject route = new JsonObject();
         route.put("destination", target);
 
+        if(queueHeader != null) {
+            route.put("headers", new JsonArray(List.of(new JsonObject().put("header", "x-queue").put("value", queueHeader))));
+        }
         if(methods != null){
             route.put("methods", new JsonArray(Arrays.asList(methods)));
         }

diff --git a/gateleen-test/src/test/java/org/swisspush/gateleen/hook/ListenerTest.java b/gateleen-test/src/test/java/org/swisspush/gateleen/hook/ListenerTest.java
@@ -11,6 +11,7 @@
 import io.vertx.ext.unit.Async;
 import io.vertx.ext.unit.TestContext;
 import io.vertx.ext.unit.junit.VertxUnitRunner;
+import org.junit.Ignore;
 import org.junit.Rule;
 import org.junit.Test;
 import org.junit.runner.RunWith;
@@ -63,6 +64,7 @@ private void initRoutingRules() {
         // add a routing
         JsonObject rules = new JsonObject();
         rules = TestUtils.addRoutingRuleMainStorage(rules);
+        rules = TestUtils.addRoutingRuleQueuing(rules);
         rules = TestUtils.addRoutingRuleHooks(rules);
         TestUtils.putRoutingRules(rules);
     }
@@ -678,7 +680,7 @@ public void testHookQueueExpiryOverride(TestContext context) {
         String requestUrl = sourceUrl + TestUtils.getHookListenersUrlSuffix() + "testservice" + "/" + 1;
         String targetUrl = targetUrlBase + "/result";
 
-        String queueName = HookHandler.LISTENER_QUEUE_PREFIX + "-" + hookHandler.getUniqueListenerId(SERVER_ROOT + requestUrl);
+        String queueName = "hook-queue-expiry-test";
 
         String putRequest = sourceUrl + "/test1";
         String putTarget = targetUrl + "/test1";
@@ -689,7 +691,8 @@ public void testHookQueueExpiryOverride(TestContext context) {
         // ----
 
         // register Listener
-        TestUtils.registerListener(requestUrl, targetUrl, null, null, 5);
+        TestUtils.registerListener(requestUrl, targetUrl, null, null, 5,
+                null, null, null, queueName);
 
         // lock queue
         String lockRequestUrl = "queuing/locks/" + queueName;

diff --git a/gateleen-test/src/test/java/org/swisspush/gateleen/queue/QueueTest.java b/gateleen-test/src/test/java/org/swisspush/gateleen/queue/QueueTest.java
@@ -30,6 +30,7 @@ public void init() {
         // add a routing
         JsonObject rules = new JsonObject();
         rules = TestUtils.addRoutingRuleMainStorage(rules);
+        rules = TestUtils.addRoutingRuleQueuing(rules);
         rules = TestUtils.addRoutingRuleCleanup(rules);
         TestUtils.putRoutingRules(rules);
     }

diff --git a/gateleen-test/src/test/java/org/swisspush/gateleen/queue/expiry/ResourceQueueExpiryTest.java b/gateleen-test/src/test/java/org/swisspush/gateleen/queue/expiry/ResourceQueueExpiryTest.java
@@ -1,6 +1,7 @@
 package org.swisspush.gateleen.queue.expiry;
 
 import io.restassured.RestAssured;
+import io.vertx.core.json.JsonObject;
 import io.vertx.ext.unit.Async;
 import io.vertx.ext.unit.TestContext;
 import io.vertx.ext.unit.junit.VertxUnitRunner;
@@ -36,6 +37,14 @@ public void initRestAssured() {
         RestAssured.requestSpecification.basePath(SERVER_ROOT + "/");
     }
 
+    private void initRoutingRules() {
+        // add a routing
+        JsonObject rules = new JsonObject();
+        rules = TestUtils.addRoutingRuleMainStorage(rules);
+        rules = TestUtils.addRoutingRuleQueuing(rules);
+        TestUtils.putRoutingRules(rules);
+    }
+
     /**
      * Checks if the GET request of the
      * given resource returns the wished body.
@@ -55,6 +64,7 @@ private void checkGETBodyWithAwait(final String requestUrl, final String body) {
     public void testQueueExpiry(TestContext context) {
         Async async = context.async();
         delete();
+        initRoutingRules();
 
         System.out.println("testQueueExpiry");
 
@@ -165,6 +175,7 @@ public void testQueueExpiry(TestContext context) {
     public void testQueueExpiryOverride_requestIsExpired_beforeRegularExpiryTime(TestContext context) {
         Async async = context.async();
         delete();
+        initRoutingRules();
 
         System.out.println("testQueueExpiry");
 
@@ -213,6 +224,7 @@ public void testQueueExpiryOverride_requestIsExpired_beforeRegularExpiryTime(Tes
     public void testQueueExpiryOverride_requestIsNotExpired_regularResourceExpiry(TestContext context) {
         Async async = context.async();
         delete();
+        initRoutingRules();
 
         System.out.println("testQueueExpiry");