Skip to content

Commit 271c89b

Browse files
Kusigrunnerguentherm
authored
Release (#568)
* updating poms for 2.1.2 branch with snapshot versions * updating poms for 2.1.3-SNAPSHOT development * updating poms for branch'release-2.1.2' with non-snapshot versions * updating develop poms to master versions to avoid merge conflicts * Updating develop poms back to pre merge state * Increase Github Actions timeout * #550 QueueSplitter for HookHandler (#566) * #550 QueueSplitter for HookHandler * #550 Increase Maven build timeout * #550 Fix initialization order and decrease timeout again * #550 Cleanup unused import --------- Co-authored-by: guentherm <[email protected]> --------- Co-authored-by: runner <runner@fv-az984-138.r1dl3knd25iuznqlwxibdlge2f.cx.internal.cloudapp.net> Co-authored-by: guentherm <[email protected]>
1 parent 2aa9d84 commit 271c89b

File tree

27 files changed

+104
-69
lines changed

27 files changed

+104
-69
lines changed

‎gateleen-cache/pom.xml

+1-1
Original file line numberDiff line numberDiff line change
@@ -6,7 +6,7 @@
66
<parent>
77
<groupId>org.swisspush.gateleen</groupId>
88
<artifactId>gateleen</artifactId>
9-
<version>2.1.2-SNAPSHOT</version>
9+
<version>2.1.3-SNAPSHOT</version>
1010
</parent>
1111

1212
<artifactId>gateleen-cache</artifactId>

‎gateleen-core/pom.xml

+1-1
Original file line numberDiff line numberDiff line change
@@ -6,7 +6,7 @@
66
<parent>
77
<groupId>org.swisspush.gateleen</groupId>
88
<artifactId>gateleen</artifactId>
9-
<version>2.1.2-SNAPSHOT</version>
9+
<version>2.1.3-SNAPSHOT</version>
1010
</parent>
1111

1212
<artifactId>gateleen-core</artifactId>

‎gateleen-delegate/pom.xml

+1-1
Original file line numberDiff line numberDiff line change
@@ -6,7 +6,7 @@
66
<parent>
77
<groupId>org.swisspush.gateleen</groupId>
88
<artifactId>gateleen</artifactId>
9-
<version>2.1.2-SNAPSHOT</version>
9+
<version>2.1.3-SNAPSHOT</version>
1010
</parent>
1111

1212
<artifactId>gateleen-delegate</artifactId>

‎gateleen-delta/pom.xml

+1-1
Original file line numberDiff line numberDiff line change
@@ -6,7 +6,7 @@
66
<parent>
77
<groupId>org.swisspush.gateleen</groupId>
88
<artifactId>gateleen</artifactId>
9-
<version>2.1.2-SNAPSHOT</version>
9+
<version>2.1.3-SNAPSHOT</version>
1010
</parent>
1111

1212
<artifactId>gateleen-delta</artifactId>

‎gateleen-expansion/pom.xml

+1-1
Original file line numberDiff line numberDiff line change
@@ -6,7 +6,7 @@
66
<parent>
77
<groupId>org.swisspush.gateleen</groupId>
88
<artifactId>gateleen</artifactId>
9-
<version>2.1.2-SNAPSHOT</version>
9+
<version>2.1.3-SNAPSHOT</version>
1010
</parent>
1111

1212
<artifactId>gateleen-expansion</artifactId>

‎gateleen-hook-js/pom.xml

+1-1
Original file line numberDiff line numberDiff line change
@@ -4,7 +4,7 @@
44
<parent>
55
<groupId>org.swisspush.gateleen</groupId>
66
<artifactId>gateleen</artifactId>
7-
<version>2.1.2-SNAPSHOT</version>
7+
<version>2.1.3-SNAPSHOT</version>
88
</parent>
99
<artifactId>gateleen-hook-js</artifactId>
1010
<packaging>jar</packaging>

‎gateleen-hook/pom.xml

+1-1
Original file line numberDiff line numberDiff line change
@@ -6,7 +6,7 @@
66
<parent>
77
<groupId>org.swisspush.gateleen</groupId>
88
<artifactId>gateleen</artifactId>
9-
<version>2.1.2-SNAPSHOT</version>
9+
<version>2.1.3-SNAPSHOT</version>
1010
</parent>
1111

1212
<artifactId>gateleen-hook</artifactId>

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

+71-31
Original file line numberDiff line numberDiff line change
@@ -10,7 +10,12 @@
1010
import io.vertx.core.Vertx;
1111
import io.vertx.core.buffer.Buffer;
1212
import io.vertx.core.eventbus.Message;
13-
import io.vertx.core.http.*;
13+
import io.vertx.core.http.HttpClient;
14+
import io.vertx.core.http.HttpClientRequest;
15+
import io.vertx.core.http.HttpClientResponse;
16+
import io.vertx.core.http.HttpMethod;
17+
import io.vertx.core.http.HttpServerRequest;
18+
import io.vertx.core.http.HttpServerResponse;
1419
import io.vertx.core.http.impl.headers.HeadersMultiMap;
1520
import io.vertx.core.json.DecodeException;
1621
import io.vertx.core.json.JsonArray;
@@ -26,8 +31,17 @@
2631
import org.swisspush.gateleen.core.logging.LoggableResource;
2732
import org.swisspush.gateleen.core.logging.RequestLogger;
2833
import org.swisspush.gateleen.core.storage.ResourceStorage;
29-
import org.swisspush.gateleen.core.util.*;
30-
import org.swisspush.gateleen.hook.queueingstrategy.*;
34+
import org.swisspush.gateleen.core.util.CollectionContentComparator;
35+
import org.swisspush.gateleen.core.util.HttpHeaderUtil;
36+
import org.swisspush.gateleen.core.util.HttpRequestHeader;
37+
import org.swisspush.gateleen.core.util.HttpServerRequestUtil;
38+
import org.swisspush.gateleen.core.util.ResourcesUtils;
39+
import org.swisspush.gateleen.core.util.StatusCode;
40+
import org.swisspush.gateleen.hook.queueingstrategy.DefaultQueueingStrategy;
41+
import org.swisspush.gateleen.hook.queueingstrategy.DiscardPayloadQueueingStrategy;
42+
import org.swisspush.gateleen.hook.queueingstrategy.QueueingStrategy;
43+
import org.swisspush.gateleen.hook.queueingstrategy.QueueingStrategyFactory;
44+
import org.swisspush.gateleen.hook.queueingstrategy.ReducedPropagationQueueingStrategy;
3145
import org.swisspush.gateleen.hook.reducedpropagation.ReducedPropagationManager;
3246
import org.swisspush.gateleen.logging.LogAppenderRepository;
3347
import org.swisspush.gateleen.logging.LoggingResourceManager;
@@ -36,13 +50,24 @@
3650
import org.swisspush.gateleen.queue.queuing.QueueClient;
3751
import org.swisspush.gateleen.queue.queuing.QueueProcessor;
3852
import org.swisspush.gateleen.queue.queuing.RequestQueue;
53+
import org.swisspush.gateleen.queue.queuing.splitter.NoOpQueueSplitter;
54+
import org.swisspush.gateleen.queue.queuing.splitter.QueueSplitter;
3955
import org.swisspush.gateleen.routing.Router;
4056
import org.swisspush.gateleen.routing.Rule;
4157
import org.swisspush.gateleen.routing.RuleFactory;
4258
import org.swisspush.gateleen.validation.RegexpValidator;
4359
import org.swisspush.gateleen.validation.ValidationException;
4460

45-
import java.util.*;
61+
import javax.annotation.Nonnull;
62+
import javax.annotation.Nullable;
63+
import java.util.ArrayList;
64+
import java.util.Comparator;
65+
import java.util.Iterator;
66+
import java.util.List;
67+
import java.util.Map;
68+
import java.util.Objects;
69+
import java.util.Optional;
70+
import java.util.Set;
4671
import java.util.concurrent.atomic.AtomicInteger;
4772
import java.util.function.Consumer;
4873
import java.util.regex.Pattern;
@@ -126,6 +151,8 @@ public class HookHandler implements LoggableResource {
126151
private final JsonSchema jsonSchemaHook;
127152
private int routeMultiplier;
128153

154+
private final QueueSplitter queueSplitter;
155+
129156

130157
/**
131158
* Creates a new HookHandler.
@@ -189,19 +216,28 @@ public HookHandler(Vertx vertx, HttpClient selfClient, final ResourceStorage sto
189216
public HookHandler(Vertx vertx, HttpClient selfClient, final ResourceStorage storage,
190217
LoggingResourceManager loggingResourceManager, LogAppenderRepository logAppenderRepository, MonitoringHandler monitoringHandler,
191218
String userProfilePath, String hookRootUri, RequestQueue requestQueue, boolean listableRoutes,
192-
ReducedPropagationManager reducedPropagationManager) {
219+
@Nullable ReducedPropagationManager reducedPropagationManager) {
193220
this(vertx, selfClient, storage, loggingResourceManager, logAppenderRepository, monitoringHandler, userProfilePath, hookRootUri,
194221
requestQueue, listableRoutes, reducedPropagationManager, null, storage);
195222
}
196223

197-
public HookHandler(Vertx vertx, HttpClient selfClient, final ResourceStorage storage,
224+
public HookHandler(Vertx vertx, HttpClient selfClient, final ResourceStorage userProfileStorage,
198225
LoggingResourceManager loggingResourceManager, LogAppenderRepository logAppenderRepository, MonitoringHandler monitoringHandler,
199226
String userProfilePath, String hookRootUri, RequestQueue requestQueue, boolean listableRoutes,
200-
ReducedPropagationManager reducedPropagationManager, Handler doneHandler, ResourceStorage hookStorage) {
201-
this(vertx, selfClient, storage, loggingResourceManager, logAppenderRepository, monitoringHandler, userProfilePath, hookRootUri,
227+
ReducedPropagationManager reducedPropagationManager, @Nullable Handler doneHandler, ResourceStorage hookStorage) {
228+
this(vertx, selfClient, userProfileStorage, loggingResourceManager, logAppenderRepository, monitoringHandler, userProfilePath, hookRootUri,
202229
requestQueue, listableRoutes, reducedPropagationManager, doneHandler, hookStorage, Router.DEFAULT_ROUTER_MULTIPLIER);
203230
}
204231

232+
public HookHandler(Vertx vertx, HttpClient selfClient, final ResourceStorage userProfileStorage,
233+
LoggingResourceManager loggingResourceManager, LogAppenderRepository logAppenderRepository, MonitoringHandler monitoringHandler,
234+
String userProfilePath, String hookRootUri, RequestQueue requestQueue, boolean listableRoutes,
235+
ReducedPropagationManager reducedPropagationManager, @Nullable Handler doneHandler, ResourceStorage hookStorage,
236+
int routeMultiplier) {
237+
this(vertx, selfClient, userProfileStorage, loggingResourceManager, logAppenderRepository, monitoringHandler, userProfilePath, hookRootUri,
238+
requestQueue, listableRoutes, reducedPropagationManager, doneHandler, hookStorage, routeMultiplier, new NoOpQueueSplitter());
239+
}
240+
205241
/**
206242
* Creates a new HookHandler.
207243
*
@@ -220,12 +256,14 @@ public HookHandler(Vertx vertx, HttpClient selfClient, final ResourceStorage sto
220256
* @param routeMultiplier the multiplier that is applied to routes, this is typically the number of nodes in
221257
* a cluster multiplied by the number of router instances within a node. Or in other words
222258
* the number of {@link Router} instances within a cluster
259+
* @param queueSplitter Configured QueueSplitter or NoOpQueueSplitter which dispatches to subqueues for
260+
* parallel operation.
223261
*/
224262
public HookHandler(Vertx vertx, HttpClient selfClient, final ResourceStorage userProfileStorage,
225263
LoggingResourceManager loggingResourceManager, LogAppenderRepository logAppenderRepository, MonitoringHandler monitoringHandler,
226264
String userProfilePath, String hookRootUri, RequestQueue requestQueue, boolean listableRoutes,
227-
ReducedPropagationManager reducedPropagationManager, Handler doneHandler, ResourceStorage hookStorage,
228-
int routeMultiplier) {
265+
ReducedPropagationManager reducedPropagationManager, @Nullable Handler doneHandler, ResourceStorage hookStorage,
266+
int routeMultiplier, @Nonnull QueueSplitter queueSplitter) {
229267
log.debug("Creating HookHandler ...");
230268
this.vertx = vertx;
231269
this.selfClient = selfClient;
@@ -244,7 +282,7 @@ public HookHandler(Vertx vertx, HttpClient selfClient, final ResourceStorage use
244282
this.doneHandler = doneHandler;
245283
this.hookStorage = hookStorage;
246284
this.routeMultiplier = routeMultiplier;
247-
285+
this.queueSplitter = queueSplitter;
248286
String hookSchema = ResourcesUtils.loadResource("gateleen_hooking_schema_hook", true);
249287
jsonSchemaHook = JsonSchemaFactory.getInstance().getSchema(hookSchema);
250288
}
@@ -605,11 +643,11 @@ private boolean createListingIfRequested(final HttpServerRequest request) {
605643

606644
request.response().headers().remove(HOOK_ROUTES_LISTED);
607645

608-
// if everything is fine, we add the listed collections to the given array
609-
if (response.statusCode() == StatusCode.OK.getStatusCode()) {
610-
if (log.isTraceEnabled()) {
611-
log.trace("createListingIfRequested > use existing array");
612-
}
646+
// if everything is fine, we add the listed collections to the given array
647+
if (response.statusCode() == StatusCode.OK.getStatusCode()) {
648+
if (log.isTraceEnabled()) {
649+
log.trace("createListingIfRequested > use existing array");
650+
}
613651

614652
response.handler(data -> {
615653
JsonObject responseObject = new JsonObject(data.toString());
@@ -731,7 +769,7 @@ private void installBodyHandler(final RoutingContext ctx, final List<Listener> l
731769
/**
732770
* Calls the passed listeners and passes the given handler to the enqueued listener requests.
733771
*
734-
* @param ctx original request context
772+
* @param ctx original request context
735773
* @param buffer buffer
736774
* @param filteredListeners all listeners which should be called
737775
* @param handler the handler, which should handle the requests
@@ -783,11 +821,13 @@ private void callListener(RoutingContext ctx, final Buffer buffer, final List<Li
783821
// if there is an x-queue header (after applying the header manipulator chain!),
784822
// then directly enqueue to this queue - else enqueue to a queue named alike this listener hook
785823
String queue = queueHeaders.get(X_QUEUE);
824+
786825
if (queue == null) {
787826
queue = LISTENER_QUEUE_PREFIX + "-" + listener.getListenerId(); // default queue name for this listener hook
788827
} else {
789828
queueHeaders.remove(X_QUEUE); // remove the "x-queue" header - otherwise we take a second turn through the queue
790829
}
830+
queue = queueSplitter.convertToSubQueue(queue, request);
791831

792832
QueueingStrategy queueingStrategy = listener.getHook().getQueueingStrategy();
793833

@@ -827,7 +867,7 @@ private void callListener(RoutingContext ctx, final Buffer buffer, final List<Li
827867
* The handler calls all listener (after), so this requests happen AFTER the original
828868
* request is performed.
829869
*
830-
* @param ctx original request context
870+
* @param ctx original request context
831871
* @param buffer buffer
832872
* @param afterListener list of listeners which should be called after the original request
833873
* @return the after handler
@@ -842,7 +882,7 @@ private Handler<Void> installAfterHandler(final RoutingContext ctx, final Buffer
842882
* The request happens BEFORE the original request is
843883
* performed.
844884
*
845-
* @param ctx original request context
885+
* @param ctx original request context
846886
* @param buffer buffer
847887
* @param beforeListener list of listeners which should be called before the original request
848888
* @param afterHandler the handler for listeners which have to be called after the original request
@@ -1586,17 +1626,17 @@ private void registerRoute(Buffer buffer) {
15861626
private boolean mustCreateNewRouteForHook(Route existingRoute, HttpHook newHook) {
15871627
HttpHook oldHook = existingRoute.getHook();
15881628
boolean same;
1589-
same = Objects.equals(oldHook.getDestination() , newHook.getDestination ());
1590-
same &= Objects.equals(oldHook.getMethods (), newHook.getMethods ());
1591-
same &= Objects.equals(oldHook.getTranslateStatus (), newHook.getTranslateStatus ());
1592-
same &= oldHook.isCollection () == newHook.isCollection () ;
1593-
same &= oldHook.isFullUrl () == newHook.isFullUrl () ;
1594-
same &= oldHook.isListable () == newHook.isListable () ;
1595-
same &= oldHook.isCollection () == newHook.isCollection () ;
1596-
same &= oldHook.isCollection () == newHook.isCollection () ;
1597-
same &= Objects.equals(oldHook.getConnectionPoolSize() , newHook.getConnectionPoolSize());
1598-
same &= Objects.equals(oldHook.getMaxWaitQueueSize() , newHook.getMaxWaitQueueSize ());
1599-
same &= Objects.equals(oldHook.getTimeout(), newHook.getTimeout ());
1629+
same = Objects.equals(oldHook.getDestination(), newHook.getDestination());
1630+
same &= Objects.equals(oldHook.getMethods(), newHook.getMethods());
1631+
same &= Objects.equals(oldHook.getTranslateStatus(), newHook.getTranslateStatus());
1632+
same &= oldHook.isCollection() == newHook.isCollection();
1633+
same &= oldHook.isFullUrl() == newHook.isFullUrl();
1634+
same &= oldHook.isListable() == newHook.isListable();
1635+
same &= oldHook.isCollection() == newHook.isCollection();
1636+
same &= oldHook.isCollection() == newHook.isCollection();
1637+
same &= Objects.equals(oldHook.getConnectionPoolSize(), newHook.getConnectionPoolSize());
1638+
same &= Objects.equals(oldHook.getMaxWaitQueueSize(), newHook.getMaxWaitQueueSize());
1639+
same &= Objects.equals(oldHook.getTimeout(), newHook.getTimeout());
16001640
same &= headersFilterPatternEquals(oldHook.getHeadersFilterPattern(), newHook.getHeadersFilterPattern());
16011641

16021642
// queueingStrategy, filter, queueExpireAfter and hookTriggerType are not relevant for Route-Hooks
@@ -1606,7 +1646,7 @@ private boolean mustCreateNewRouteForHook(Route existingRoute, HttpHook newHook)
16061646
}
16071647

16081648
private boolean headersFilterPatternEquals(Pattern headersFilterPatternLeft, Pattern headersFilterPatternRight) {
1609-
if(headersFilterPatternLeft != null && headersFilterPatternRight != null){
1649+
if (headersFilterPatternLeft != null && headersFilterPatternRight != null) {
16101650
return Objects.equals(headersFilterPatternLeft.pattern(), headersFilterPatternRight.pattern());
16111651
}
16121652

‎gateleen-kafka/pom.xml

+1-1
Original file line numberDiff line numberDiff line change
@@ -6,7 +6,7 @@
66
<parent>
77
<groupId>org.swisspush.gateleen</groupId>
88
<artifactId>gateleen</artifactId>
9-
<version>2.1.2-SNAPSHOT</version>
9+
<version>2.1.3-SNAPSHOT</version>
1010
</parent>
1111

1212
<artifactId>gateleen-kafka</artifactId>

‎gateleen-logging/pom.xml

+1-1
Original file line numberDiff line numberDiff line change
@@ -6,7 +6,7 @@
66
<parent>
77
<groupId>org.swisspush.gateleen</groupId>
88
<artifactId>gateleen</artifactId>
9-
<version>2.1.2-SNAPSHOT</version>
9+
<version>2.1.3-SNAPSHOT</version>
1010
</parent>
1111

1212
<artifactId>gateleen-logging</artifactId>

‎gateleen-merge/pom.xml

+1-1
Original file line numberDiff line numberDiff line change
@@ -6,7 +6,7 @@
66
<parent>
77
<groupId>org.swisspush.gateleen</groupId>
88
<artifactId>gateleen</artifactId>
9-
<version>2.1.2-SNAPSHOT</version>
9+
<version>2.1.3-SNAPSHOT</version>
1010
</parent>
1111

1212
<artifactId>gateleen-merge</artifactId>

‎gateleen-monitoring/pom.xml

+1-1
Original file line numberDiff line numberDiff line change
@@ -6,7 +6,7 @@
66
<parent>
77
<groupId>org.swisspush.gateleen</groupId>
88
<artifactId>gateleen</artifactId>
9-
<version>2.1.2-SNAPSHOT</version>
9+
<version>2.1.3-SNAPSHOT</version>
1010
</parent>
1111

1212
<artifactId>gateleen-monitoring</artifactId>

‎gateleen-packing/pom.xml

+1-1
Original file line numberDiff line numberDiff line change
@@ -6,7 +6,7 @@
66
<parent>
77
<groupId>org.swisspush.gateleen</groupId>
88
<artifactId>gateleen</artifactId>
9-
<version>2.1.2-SNAPSHOT</version>
9+
<version>2.1.3-SNAPSHOT</version>
1010
</parent>
1111

1212
<artifactId>gateleen-packing</artifactId>

‎gateleen-player/pom.xml

+1-1
Original file line numberDiff line numberDiff line change
@@ -6,7 +6,7 @@
66
<parent>
77
<groupId>org.swisspush.gateleen</groupId>
88
<artifactId>gateleen</artifactId>
9-
<version>2.1.2-SNAPSHOT</version>
9+
<version>2.1.3-SNAPSHOT</version>
1010
</parent>
1111

1212
<artifactId>gateleen-player</artifactId>

‎gateleen-playground/pom.xml

+1-1
Original file line numberDiff line numberDiff line change
@@ -6,7 +6,7 @@
66
<parent>
77
<groupId>org.swisspush.gateleen</groupId>
88
<artifactId>gateleen</artifactId>
9-
<version>2.1.2-SNAPSHOT</version>
9+
<version>2.1.3-SNAPSHOT</version>
1010
</parent>
1111

1212
<artifactId>gateleen-playground</artifactId>

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

+4-4
Original file line numberDiff line numberDiff line change
@@ -257,9 +257,12 @@ public void start() {
257257
queueClient, lock);
258258
reducedPropagationManager.startExpiredQueueProcessing(5000);
259259

260+
queueSplitter = new QueueSplitterImpl(configurationResourceManager, SERVER_ROOT + "/admin/v1/queueSplitters");
261+
queueSplitter.initialize();
262+
260263
hookHandler = new HookHandler(vertx, selfClient, storage, loggingResourceManager, logAppenderRepository,
261264
monitoringHandler,SERVER_ROOT + "/users/v1/%s/profile",
262-
SERVER_ROOT + "/hooks/v1/", queueClient,false, reducedPropagationManager);
265+
SERVER_ROOT + "/hooks/v1/", queueClient,false, reducedPropagationManager, null, storage, Router.DEFAULT_ROUTER_MULTIPLIER, queueSplitter);
263266
hookHandler.enableResourceLogging(true);
264267

265268
authorizer = new Authorizer(vertx, storage, SERVER_ROOT + "/security/v1/", ROLE_PATTERN, ROLE_PREFIX, props);
@@ -328,9 +331,6 @@ public void start() {
328331
final QueueBrowser queueBrowser = new QueueBrowser(vertx, SERVER_ROOT + "/queuing", Address.redisquesAddress(),
329332
monitoringHandler);
330333

331-
queueSplitter = new QueueSplitterImpl(configurationResourceManager, SERVER_ROOT + "/admin/v1/queueSplitters");
332-
queueSplitter.initialize();
333-
334334
LogController logController = new LogController();
335335
logController.registerLogConfiguratorMBean(JMX_DOMAIN);
336336

‎gateleen-qos/pom.xml

+1-1
Original file line numberDiff line numberDiff line change
@@ -6,7 +6,7 @@
66
<parent>
77
<groupId>org.swisspush.gateleen</groupId>
88
<artifactId>gateleen</artifactId>
9-
<version>2.1.2-SNAPSHOT</version>
9+
<version>2.1.3-SNAPSHOT</version>
1010
</parent>
1111

1212
<artifactId>gateleen-qos</artifactId>

‎gateleen-queue/pom.xml

+1-1
Original file line numberDiff line numberDiff line change
@@ -6,7 +6,7 @@
66
<parent>
77
<groupId>org.swisspush.gateleen</groupId>
88
<artifactId>gateleen</artifactId>
9-
<version>2.1.2-SNAPSHOT</version>
9+
<version>2.1.3-SNAPSHOT</version>
1010
</parent>
1111

1212
<artifactId>gateleen-queue</artifactId>

0 commit comments

Comments
 (0)