#186 Enqueue directly to destination queue

Author: Thomas Winterhalder IT132

.gitignore

@@ -7,4 +7,7 @@ target
 .vertx
 build
 .gradle
-playground_logs.log
+playground_logs.log
+**/node/
+**/node_modules/
+redis

gateleen-hook/src/main/java/org/swisspush/gateleen/hook/HookHandler.java

@@ -45,6 +45,7 @@ public class HookHandler implements LoggableResource {
     public static final String HOOK_ROUTES_LISTED = "x-hook-routes-listed";
     public static final String HOOKS_LISTENERS_URI_PART = "/_hooks/listeners/";
     public static final String LISTENER_QUEUE_PREFIX = "listener-hook";
+    private static final String X_QUEUE = "x-queue";
     private static final String LISTENER_HOOK_TARGET_PATH = "listeners/";
 
     public static final String HOOKS_ROUTE_URI_PART = "/_hooks/route";
@@ -683,7 +684,7 @@ private void callListener(final HttpServerRequest request, final Buffer buffer,
                 log.debug(" > external target: " + targetUri);
             }
 
-            String queue = LISTENER_QUEUE_PREFIX + "-" + listener.getListenerId();
+            String queue = listener.getDestinationQueue();
 
             // Create a new multimap, copied from the original request,
             // so that the original request is not overridden with the new values.
@@ -1232,6 +1233,16 @@ private void registerListener(Buffer buffer) {
 
         extractAndAddStaticHeadersToHook(jsonHook, hook);
 
+        Map<String, String> staticHeaders = hook.getStaticHeaders();
+        String destinationQueue = null;
+        if (staticHeaders != null) {
+            // remove static x-queue header. Otherwise it will be re-queued at the end of the queue.
+            destinationQueue = hook.getStaticHeaders().remove(X_QUEUE);
+        }
+        if (destinationQueue == null) {
+            destinationQueue = LISTENER_QUEUE_PREFIX + "-" + listenerId;
+        }
+
         /*
          * Despite the fact, that every hook
          * should have an expiration time,
@@ -1281,7 +1292,7 @@ private void registerListener(Buffer buffer) {
         }
 
         // create and add a new listener (or update an already existing listener)
-        listenerRepository.addListener(new Listener(listenerId, getMonitoredUrlSegment(requestUrl), target, hook));
+        listenerRepository.addListener(new Listener(listenerId, getMonitoredUrlSegment(requestUrl), target, hook, destinationQueue));
     }
 
     /**
@@ -1296,7 +1307,7 @@ private void extractAndAddStaticHeadersToHook(final JsonObject jsonHook, final H
         if (staticHeaders != null && staticHeaders.size() > 0) {
             hook.addStaticHeaders(new LinkedHashMap<>());
             for (Map.Entry<String, Object> entry : staticHeaders.getMap().entrySet()) {
-                hook.getStaticHeaders().put(entry.getKey(), entry.getValue().toString());
+                hook.getStaticHeaders().put(entry.getKey().toLowerCase(), entry.getValue().toString());
             }
         }
     }

gateleen-hook/src/main/java/org/swisspush/gateleen/hook/Listener.java

@@ -11,6 +11,7 @@ public class Listener {
     private String monitoredUrl;
     private Integer expireAfter;
     private HttpHook hook;
+    private String destinationQueue;
 
     /**
      * Creates a new instance of a HookListener.
@@ -26,11 +27,12 @@ public class Listener {
      * If it's a local listener, this can also be the target url.
      * @param hook - the hook of this listener
      */
-    public Listener(String listenerId, String monitoredUrl, String listener, HttpHook hook) {
+    public Listener(String listenerId, String monitoredUrl, String listener, HttpHook hook, String destinationQueue) {
         this.listenerId = listenerId;
         this.monitoredUrl = monitoredUrl;
         this.listener = listener;
         this.setHook(hook);
+        this.setDestinationQueue(destinationQueue);
     }
 
     /**
@@ -127,4 +129,21 @@ public void setHook(HttpHook hook) {
         this.hook = hook;
     }
 
+    /**
+     * Returns the destinationQueue of this listener.
+     *
+     * @return String
+     */
+    public String getDestinationQueue() {
+        return destinationQueue;
+    }
+
+    /**
+     * Sets the destinationQueue for this listener.
+     *
+     * @param destinationQueue queue name
+     */
+    public void setDestinationQueue(String destinationQueue) {
+        this.destinationQueue = destinationQueue;
+    }
 }