Merge pull request #315 from swisspost/develop

PR for release

Author: mcweba

pom.xml

@@ -2,7 +2,7 @@
     <modelVersion>4.0.0</modelVersion>
     <groupId>org.swisspush</groupId>
     <artifactId>redisques</artifactId>
-    <version>4.1.21-SNAPSHOT</version>
+    <version>4.1.22-SNAPSHOT</version>
     <name>redisques</name>
     <description>
         A highly scalable redis-persistent queuing system for vertx

src/main/java/org/swisspush/redisques/handler/RedisquesHttpRequestHandler.java

@@ -85,8 +85,13 @@ public static void init(
             if (modConfig.getHttpRequestHandlerPort() != null && modConfig.getHttpRequestHandlerUserHeader() != null) {
                 var handler = new RedisquesHttpRequestHandler(
                         vertx, modConfig, queueStatsService, exceptionFactory);
+
                 // in Vert.x 2x 100-continues was activated per default, in vert.x 3x it is off per default.
-                HttpServerOptions options = new HttpServerOptions().setHandle100ContinueAutomatically(true);
+                HttpServerOptions options = new HttpServerOptions()
+                        .setHandle100ContinueAutomatically(true)
+                        .setMaxHeaderSize(modConfig.getHttpRequestHandlerMaxHeaderSize())
+                        .setMaxInitialLineLength(modConfig.getHttpRequestHandlerMaxInitialLineLength());
+
                 vertx.createHttpServer(options).requestHandler(handler).listen(modConfig.getHttpRequestHandlerPort(), result -> {
                     if (result.succeeded()) {
                         log.info("Successfully started http request handler on port {}", modConfig.getHttpRequestHandlerPort());

src/main/java/org/swisspush/redisques/queue/QueueConsumerRunner.java

@@ -1,6 +1,5 @@
 package org.swisspush.redisques.queue;
 
-import com.google.common.math.PairedStats;
 import io.vertx.core.AsyncResult;
 import io.vertx.core.Future;
 import io.vertx.core.Handler;
@@ -32,6 +31,8 @@
 import java.util.function.Function;
 import java.util.stream.Collectors;
 
+import static io.vertx.core.Future.failedFuture;
+import static io.vertx.core.Future.succeededFuture;
 import static java.lang.System.currentTimeMillis;
 import static org.swisspush.redisques.util.RedisquesAPI.OK;
 import static org.swisspush.redisques.util.RedisquesAPI.PAYLOAD;
@@ -359,16 +360,13 @@ private void processMessageWithTimeout(final String queue, final String payload,
     }
 
     private Future<Void> refreshRegistration(String queueName) {
-        Promise<Void> promise = Promise.promise();
-        vertx.eventBus().request(keyspaceHelper.getVerticleRefreshRegistrationKey(), queueName).onComplete(event -> {
-            if (event.succeeded()) {
-                promise.complete();
-                return;
+        String address = keyspaceHelper.getVerticleRefreshRegistrationKey();
+        return vertx.eventBus().request(address, queueName).transform(ev -> {
+            if (ev.failed()) {
+                log.error("RedisQues refresh registration failed", ev.cause());
             }
-            promise.fail(event.cause());
-            log.error("RedisQues refresh registration failed", event.cause());
+            return ev.succeeded() ? succeededFuture() : failedFuture(ev.cause());
         });
-        return promise.future();
     }
 
     private Future<Void> notifyConsumer(String queueName) {
@@ -422,15 +420,13 @@ void setMyQueuesState(String queueName, QueueState state) {
             if (null == queueProcessingState) {
                 // not in our list yet
                 return new QueueProcessingState(state, 0);
+            } else if (queueProcessingState.getState() == QueueState.CONSUMING && state == QueueState.READY) {
+                // update the state and the timestamp when we change from CONSUMING to READY
+                return new QueueProcessingState(QueueState.READY, currentTimeMillis());
             } else {
-                if (queueProcessingState.getState() == QueueState.CONSUMING && state == QueueState.READY) {
-                    // update the state and the timestamp when we change from CONSUMING to READY
-                    return new QueueProcessingState(QueueState.READY, currentTimeMillis());
-                } else {
-                    // update the state but leave the timestamp unchanged
-                    queueProcessingState.setState(state);
-                    return queueProcessingState;
-                }
+                // update the state but leave the timestamp unchanged
+                queueProcessingState.setState(state);
+                return queueProcessingState;
             }
         });
     }

src/main/java/org/swisspush/redisques/queue/QueueRegistryService.java

@@ -228,12 +228,13 @@ public void refreshRegistration(String queueName, Handler<AsyncResult<Response>>
         if (handler == null) {
             throw new RuntimeException("Handler must be set");
         } else {
-            vertx.executeBlocking(promise ->
-                    redisService.expire(consumerKey, String.valueOf(consumerLockTime)).
-                            onComplete(responseAsyncResult -> {
-                                handler.handle(responseAsyncResult);
-                                promise.complete();
-                            }));
+            vertx.executeBlocking(() -> {
+                return redisService.expire(consumerKey, String.valueOf(consumerLockTime));
+            }).compose((Future<Response> tooManyNestedFutures) -> {
+                return tooManyNestedFutures;
+            }).onComplete((AsyncResult<Response> ev) -> {
+                handler.handle(ev);
+            });
         }
     }
 

src/main/java/org/swisspush/redisques/queue/RedisService.java

@@ -81,19 +81,16 @@ public Future<Response> get(String key) {
     }
 
     public Future<Boolean> setNxPx(String key, String value, boolean setNx, long durationMillis) {
-        Promise<Boolean> p = Promise.promise();
-        redis().compose((RedisAPI redisAPI) -> {
-                    Future<?> future;
-                    if (setNx) {
-                        future = redisAPI.send(Command.SET, key, value, "NX", "PX", String.valueOf(durationMillis));
-                    } else {
-                        future = redisAPI.send(Command.SET, key, value, "PX", String.valueOf(durationMillis));
-                    }
-                    return future.onSuccess(resp -> p.complete(resp != null && "OK".equals(resp.toString())))
-                            .onFailure(p::fail);
-                }
-        );
-        return p.future();
+        return redis().compose((RedisAPI redisAPI) -> {
+            String durationMillisStr = String.valueOf(durationMillis);
+            if (setNx) {
+                return redisAPI.send(Command.SET, key, value, "NX", "PX", durationMillisStr);
+            } else {
+                return redisAPI.send(Command.SET, key, value, "PX", durationMillisStr);
+            }
+        }).map((Response rsp) -> {
+            return rsp != null && "OK".equals(rsp.toString());
+        });
     }
 
     public Future<Response> zadd(String queuesKey, String queueName, String value) {