Feature/550 queue splitter

gateleen-playground/src/main/java/org/swisspush/gateleen/playground/Server.java

@@ -64,6 +64,8 @@
 import org.swisspush.gateleen.queue.queuing.circuitbreaker.impl.QueueCircuitBreakerImpl;
 import org.swisspush.gateleen.queue.queuing.circuitbreaker.impl.RedisQueueCircuitBreakerStorage;
 import org.swisspush.gateleen.queue.queuing.circuitbreaker.util.QueueCircuitBreakerRulePatternToCircuitMapping;
+import org.swisspush.gateleen.queue.queuing.splitter.QueueSplitter;
+import org.swisspush.gateleen.queue.queuing.splitter.QueueSplitterImpl;
 import org.swisspush.gateleen.routing.CustomHttpResponseHandler;
 import org.swisspush.gateleen.routing.DeferCloseHttpClient;
 import org.swisspush.gateleen.routing.Router;
@@ -154,6 +156,8 @@ public class Server extends AbstractVerticle {
     private ContentTypeConstraintHandler contentTypeConstraintHandler;
     private CacheHandler cacheHandler;
 
+    private QueueSplitter queueSplitter;
+
     public static void main(String[] args) {
         Vertx.vertx().deployVerticle("org.swisspush.gateleen.playground.Server", event ->
                 LoggerFactory.getLogger(Server.class).info("[_] Gateleen - http://localhost:7012/gateleen/")
@@ -323,6 +327,9 @@ public void start() {
                 final QueueBrowser queueBrowser = new QueueBrowser(vertx, SERVER_ROOT + "/queuing", Address.redisquesAddress(),
                         monitoringHandler);
 
+                queueSplitter = new QueueSplitterImpl(configurationResourceManager, SERVER_ROOT + "/admin/v1/queueSplitters");
+                queueSplitter.initialize();
+
                 LogController logController = new LogController();
                 logController.registerLogConfiguratorMBean(JMX_DOMAIN);
 
@@ -347,6 +354,7 @@ public void start() {
                         .loggingResourceManager(loggingResourceManager)
                         .configurationResourceManager(configurationResourceManager)
                         .queueCircuitBreakerConfigurationResourceManager(queueCircuitBreakerConfigurationResourceManager)
+                        .queueSplitter(queueSplitter)
                         .schedulerResourceManager(schedulerResourceManager)
                         .zipExtractHandler(zipExtractHandler)
                         .delegateHandler(delegateHandler)

gateleen-playground/src/main/resources/playground/server/admin/v1/queueSplitters

@@ -0,0 +1,33 @@
+{
+    "queue-static-split": {
+        "description": "Simple splitter with request header",
+        "postfixFromStatic": [
+            "A",
+            "B",
+            "C",
+            "D"
+        ]
+    },
+    "queue-header-[a-z]+": {
+        "description": "Simple splitter with request header",
+        "postfixDelimiter": "+",
+        "postfixFromRequest": {
+            "header": "x-rp-deviceid"
+        }
+    },
+    "queue-path-[a-z]+": {
+        "description": "Simple splitter with request url matching",
+        "postfixDelimiter": "_",
+        "postfixFromRequest": {
+            "url": ".*/path1/(.*)/.*"
+        }
+    },
+    "queue-header-and-path-[a-z]+": {
+        "description": "Simple splitter with request header and url matching",
+        "postfixDelimiter": "_",
+        "postfixFromRequest": {
+            "header": "x-rp-deviceid",
+            "url": ".*/path1/(.*)/.*"
+        }
+    }
+}

gateleen-queue/src/main/java/org/swisspush/gateleen/queue/queuing/QueuingHandler.java

@@ -11,6 +11,8 @@
 import org.swisspush.gateleen.core.util.StatusCode;
 import org.swisspush.gateleen.monitoring.MonitoringHandler;
 import org.swisspush.gateleen.queue.duplicate.DuplicateCheckHandler;
+import org.swisspush.gateleen.queue.queuing.splitter.NoOpQueueSplitter;
+import org.swisspush.gateleen.queue.queuing.splitter.QueueSplitter;
 
 import static org.swisspush.redisques.util.RedisquesAPI.buildCheckOperation;
 
@@ -34,16 +36,45 @@
     private final HttpServerRequest request;
     private final Vertx vertx;
     private final RedisProvider redisProvider;
+    private final QueueSplitter queueSplitter;
 
-    public QueuingHandler(Vertx vertx, RedisProvider redisProvider, HttpServerRequest request, MonitoringHandler monitoringHandler) {
-        this(vertx, redisProvider, request, new QueueClient(vertx, monitoringHandler));
+    public QueuingHandler(
+            Vertx vertx,
+            RedisProvider redisProvider,
+            HttpServerRequest request,
+            MonitoringHandler monitoringHandler
+    ) {
+        this(vertx, redisProvider, request, new QueueClient(vertx, monitoringHandler), new NoOpQueueSplitter());
     }
 
-    public QueuingHandler(Vertx vertx, RedisProvider redisProvider, HttpServerRequest request, RequestQueue requestQueue) {
+    public QueuingHandler(
+            Vertx vertx,
+            RedisProvider redisProvider,
+            HttpServerRequest request,
+            MonitoringHandler monitoringHandler,
+            QueueSplitter queueSplitter
+    ) {
+        this(
+                vertx,
+                redisProvider,
+                request,
+                new QueueClient(vertx, monitoringHandler),
+                queueSplitter == null ? new NoOpQueueSplitter() : queueSplitter
+        );
+    }
+
+    public QueuingHandler(
+            Vertx vertx,
+            RedisProvider redisProvider,
+            HttpServerRequest request,
+            RequestQueue requestQueue,
+            QueueSplitter queueSplitter
+    ) {
         this.request = request;
         this.vertx = vertx;
         this.redisProvider = redisProvider;
         this.requestQueue = requestQueue;
+        this.queueSplitter = queueSplitter;
     }
 
     @Override
@@ -61,12 +92,12 @@
                     request.response().setStatusMessage(StatusCode.ACCEPTED.getStatusMessage());
                     request.response().end();
                 } else {
-                    requestQueue.enqueue(request, headers, buffer, queue);
+                    requestQueue.enqueue(request, headers, buffer, queueSplitter.convertToSubQueue(queue, request));
                 }
             });
 
         } else {
-            requestQueue.enqueue(request, headers, buffer, queue);
+            requestQueue.enqueue(request, headers, buffer, queueSplitter.convertToSubQueue(queue, request));
         }
     }
 

gateleen-queue/src/main/java/org/swisspush/gateleen/queue/queuing/splitter/NoOpQueueSplitter.java

@@ -0,0 +1,26 @@
+package org.swisspush.gateleen.queue.queuing.splitter;
+
+import io.vertx.core.Future;
+import io.vertx.core.Promise;
+import io.vertx.core.http.HttpServerRequest;
+
+/**
+ * {@inheritDoc}
+ */
+public class NoOpQueueSplitter implements QueueSplitter {
+
+    @Override
+    public Future<Void> initialize() {
+        Promise<Void> promise = Promise.promise();
+        promise.complete();
+        return promise.future();
+    }
+
+    @Override
+    /**
+     * {@inheritDoc}
+     */
+    public String convertToSubQueue(String queue, HttpServerRequest request) {
+        return queue;
+    }
+}

gateleen-queue/src/main/java/org/swisspush/gateleen/queue/queuing/splitter/QueueSplitter.java

@@ -0,0 +1,24 @@
+package org.swisspush.gateleen.queue.queuing.splitter;
+
+import io.vertx.core.Future;
+import io.vertx.core.http.HttpServerRequest;
+
+/**
+ * Interface for queues configured to be split in sub-queues. The method {@link QueueSplitter#convertToSubQueue(String, HttpServerRequest)}
+ * evaluates the convert of the queue name in a sub-queue name.
+ *
+ * @author https://github.com/gcastaldi [Giannandrea Castaldi]
+ */
+public interface QueueSplitter {
+
+    public Future<Void> initialize();
+
+        /**
+         * Convert the queue name in a sub-queue name. If not necessary maintains the initial queue name.
+         *
+         * @param queue
+         * @param request
+         * @return sub-queue name
+         */
+    String convertToSubQueue(String queue, HttpServerRequest request);
+}

gateleen-queue/src/main/java/org/swisspush/gateleen/queue/queuing/splitter/QueueSplitterConfiguration.java

@@ -0,0 +1,101 @@
+package org.swisspush.gateleen.queue.queuing.splitter;
+
+import javax.annotation.Nullable;
+import java.util.List;
+import java.util.Objects;
+import java.util.regex.Pattern;
+
+/**
+ * Container holding configuration values for {@link QueueSplitterImpl} identified
+ * by a queue pattern.
+ *
+ * @author https://github.com/gcastaldi [Giannandrea Castaldi]
+ */
+public class QueueSplitterConfiguration {
+
+    private final Pattern queue;
+
+    private final String postfixDelimiter;
+
+    @Nullable
+    private final List<String> postfixFromStatic;
+
+    @Nullable
+    private final String postfixFromHeader;
+
+    @Nullable
+    private final Pattern postfixFromUrl;
+
+
+    public QueueSplitterConfiguration(
+            Pattern queue,
+            String postfixDelimiter,
+            @Nullable List<String> postfixFromStatic,
+            @Nullable String postfixFromHeader,
+            @Nullable Pattern postfixFromUrl) {
+        this.queue = queue;
+        this.postfixDelimiter = postfixDelimiter;
+        this.postfixFromStatic = postfixFromStatic;
+        this.postfixFromHeader = postfixFromHeader;
+        this.postfixFromUrl = postfixFromUrl;
+    }
+
+    public Pattern getQueue() {
+        return queue;
+    }
+
+    public String getPostfixDelimiter() {
+        return postfixDelimiter;
+    }
+
+    @Nullable
+    public List<String> getPostfixFromStatic() {
+        return postfixFromStatic;
+    }
+
+    @Nullable
+    public String getPostfixFromHeader() {
+        return postfixFromHeader;
+    }
+
+    @Nullable
+    public Pattern getPostfixFromUrl() {
+        return postfixFromUrl;
+    }
+
+    @Override
+    public boolean equals(Object o) {
+        if (this == o) return true;
+        if (o == null || getClass() != o.getClass()) return false;
+        QueueSplitterConfiguration that = (QueueSplitterConfiguration) o;
+        return Objects.equals(queue, that.queue) &&
+                Objects.equals(postfixDelimiter, that.postfixDelimiter) &&
+                Objects.equals(postfixFromStatic, that.postfixFromStatic) &&
+                Objects.equals(postfixFromHeader, that.postfixFromHeader) &&
+                Objects.equals(postfixFromUrl, that.postfixFromUrl);
+    }
+
+    @Override
+    public int hashCode() {
+        return Objects.hash(queue, postfixDelimiter, postfixFromStatic, postfixFromHeader, postfixFromUrl);
+    }
+
+    @Override
+    public String toString() {
+        return "QueueSplitterConfiguration{" +
+                "queue=" + queue +
+                ", postfixDelimiter='" + postfixDelimiter + '\'' +
+                ", postfixFromStatic=" + postfixFromStatic +
+                ", postfixFromHeader='" + postfixFromHeader + '\'' +
+                ", postfixFromUrl='" + postfixFromUrl + '\'' +
+                '}';
+    }
+
+    public boolean isSplitStatic() {
+        return postfixFromStatic != null && !postfixFromStatic.isEmpty();
+    }
+
+    public boolean isSplitFromRequest() {
+        return postfixFromHeader != null || postfixFromUrl != null;
+    }
+}

gateleen-queue/src/main/java/org/swisspush/gateleen/queue/queuing/splitter/QueueSplitterConfigurationParser.java

@@ -0,0 +1,87 @@
+package org.swisspush.gateleen.queue.queuing.splitter;
+
+import io.vertx.core.buffer.Buffer;
+import io.vertx.core.json.JsonArray;
+import io.vertx.core.json.JsonObject;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.swisspush.gateleen.core.util.StringUtils;
+
+import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.regex.Pattern;
+import java.util.regex.PatternSyntaxException;
+import java.util.stream.Collectors;
+
+/**
+ * Parses the splitter configuration resource in to a list of {@link QueueSplitterConfiguration}
+ *
+ * @author https://github.com/gcastaldi [Giannandrea Castaldi]
+ */
+public class QueueSplitterConfigurationParser {
+
+    private static final Logger log = LoggerFactory.getLogger(QueueSplitterConfigurationParser.class);
+    public static final String POSTFIX_FROM_STATIC_KEY = "postfixFromStatic";
+    public static final String POSTFIX_FROM_REQUEST_KEY = "postfixFromRequest";
+    public static final String POSTFIX_FROM_HEADER_KEY = "header";
+    public static final String POSTFIX_FROM_URL_KEY = "url";
+    public static final String POSTFIX_DELIMITER_KEY = "postfixDelimiter";
+    public static final String DEFAULT_POSTFIX_DELIMITER = "-";
+
+    static List<QueueSplitterConfiguration> parse(Buffer configurationResourceBuffer, Map<String, Object> properties) {
+
+        JsonObject config;
+        List<QueueSplitterConfiguration> queueSplitterConfigurations = new ArrayList<>();
+        try {
+            String resolvedConfiguration = StringUtils.replaceWildcardConfigs(
+                    configurationResourceBuffer.toString(StandardCharsets.UTF_8),
+                    properties
+            );
+            config = new JsonObject(Buffer.buffer(resolvedConfiguration));
+        } catch (Exception ex) {
+            log.warn("Could not replace wildcards with environment properties for queue splitter configurations or json invalid. Here the reason: {}",
+                    ex.getMessage());
+            return queueSplitterConfigurations;
+        }
+
+        for (String queuePattern : config.fieldNames()) {
+            try {
+                Pattern pattern = Pattern.compile(queuePattern);
+                JsonObject queueConfig = config.getJsonObject(queuePattern);
+                JsonArray postfixFromStatic = queueConfig.getJsonArray(POSTFIX_FROM_STATIC_KEY);
+                if (postfixFromStatic != null) {
+                    List<String> staticPostfixes = postfixFromStatic.stream().map(Object::toString).collect(Collectors.toList());
+                    queueSplitterConfigurations.add(new QueueSplitterConfiguration(
+                            pattern,
+                            queueConfig.getString("postfixDelimiter", DEFAULT_POSTFIX_DELIMITER),
+                            staticPostfixes,
+                            null,
+                            null
+                    ));
+                } else {
+                    JsonObject postfixFromRequest = queueConfig.getJsonObject(POSTFIX_FROM_REQUEST_KEY);
+                    String postfixFromHeader = postfixFromRequest != null ? postfixFromRequest.getString(POSTFIX_FROM_HEADER_KEY) : null;
+                    String postfixFromUrl = postfixFromRequest != null ? postfixFromRequest.getString(POSTFIX_FROM_URL_KEY) : null;
+                    if (postfixFromRequest != null && (postfixFromHeader != null || postfixFromUrl != null)) {
+                        queueSplitterConfigurations.add(new QueueSplitterConfiguration(
+                                pattern,
+                                queueConfig.getString(POSTFIX_DELIMITER_KEY, DEFAULT_POSTFIX_DELIMITER),
+                                null,
+                                postfixFromHeader,
+                                postfixFromUrl != null ? Pattern.compile(postfixFromUrl) : null
+                        ));
+                    } else {
+                        log.warn("Queue splitter configuration without a postfix definition");
+                    }
+                }
+            } catch (PatternSyntaxException patternException) {
+                log.warn("Queue splitter '{}' is not a valid regex pattern. Discarding this queue splitter configuration", queuePattern);
+            }
+        }
+
+        return queueSplitterConfigurations;
+    }
+
+}