HookHandler implements LoggableResource { public boolean handle(final HttpServerRequest request) { boolean consumed = false; if (isHookListenerRegistration(request)) { handleListenerRegistration(request); return true; } if (isHookListenerUnregistration(request)) { handleListenerUnregistration(request); return true; } if (isHookRouteRegistration(request)) { handleRouteRegistration(request); return true; } if (isHookRouteUnregistration(request)) { handleRouteUnregistration(request); return true; } final List<Listener> listeners = listenerRepository.findListeners(request.uri(), request.method().name()); if (!listeners.isEmpty() && !isRequestAlreadyHooked(request)) { installBodyHandler(request, listeners); consumed = true; } if (!consumed) { consumed = routeRequestIfNeeded(request); if (!consumed) { return createListingIfRequested(request); } return consumed; } else { return true; } } HookHandler(Vertx vertx, HttpClient selfClient, final ResourceStorage storage,
                       LoggingResourceManager loggingResourceManager, MonitoringHandler monitoringHandler,
                       String userProfilePath, String hookRootUri);  HookHandler(Vertx vertx, HttpClient selfClient, final ResourceStorage storage,
                       LoggingResourceManager loggingResourceManager, MonitoringHandler monitoringHandler,
                       String userProfilePath, String hookRootUri, RequestQueue requestQueue);  HookHandler(Vertx vertx, HttpClient selfClient, final ResourceStorage storage,
                       LoggingResourceManager loggingResourceManager, MonitoringHandler monitoringHandler,
                       String userProfilePath, String hookRootUri, RequestQueue requestQueue, boolean listableRoutes);  HookHandler(Vertx vertx, HttpClient selfClient, final ResourceStorage storage,
                       LoggingResourceManager loggingResourceManager, MonitoringHandler monitoringHandler,
                       String userProfilePath, String hookRootUri, RequestQueue requestQueue, boolean listableRoutes,
                       ReducedPropagationManager reducedPropagationManager);  HookHandler(Vertx vertx, HttpClient selfClient, final ResourceStorage userProfileStorage,
                       LoggingResourceManager loggingResourceManager, MonitoringHandler monitoringHandler,
                       String userProfilePath, String hookRootUri, RequestQueue requestQueue, boolean listableRoutes,
                       ReducedPropagationManager reducedPropagationManager, Handler doneHandler, ResourceStorage hookStorage); void init(); @Override void enableResourceLogging(boolean resourceLoggingEnabled); void registerListenerRegistrationHandler(Handler<Void> readyHandler); boolean handle(final HttpServerRequest request); boolean isHookJsonInvalid(HttpServerRequest request, Buffer hookData); boolean isRequestAlreadyHooked(HttpServerRequest request); static final String HOOKED_HEADER; static final String HOOK_ROUTES_LISTED; static final String HOOKS_LISTENERS_URI_PART; static final String LISTENER_QUEUE_PREFIX; static final String HOOKS_ROUTE_URI_PART; static final String REQUESTURL; static final String EXPIRATION_TIME; static final String HOOK; static final String TRANSLATE_STATUS; static final String METHODS; static final String DESTINATION; static final String FILTER; static final String QUEUE_EXPIRE_AFTER; static final String STATIC_HEADERS; static final String FULL_URL; static final String DISCARD_PAYLOAD; static final String HOOK_TRIGGER_TYPE; static final String LISTABLE; static final String COLLECTION; }
@Test public void testListenerEnqueueWithDefaultQueueingStrategy(TestContext context) throws InterruptedException { setListenerStorageEntryAndTriggerUpdate(buildListenerConfig(null, "x99")); Thread.sleep(1000); String uri = "/playground/server/tests/hooktest/abc123"; String originalPayload = "{\"key\":123}"; PUTRequest putRequest = new PUTRequest(uri, originalPayload); putRequest.addHeader(CONTENT_LENGTH.getName(), "99"); hookHandler.handle(putRequest); Mockito.verify(requestQueue, Mockito.timeout(2000).times(1)).enqueue(Mockito.argThat(new ArgumentMatcher<HttpRequest>() { @Override public boolean matches(Object argument) { HttpRequest req = (HttpRequest) argument; return HttpMethod.PUT == req.getMethod() && req.getUri().contains(uri) && new Integer(99).equals(getInteger(req.getHeaders(), CONTENT_LENGTH)) && Arrays.equals(req.getPayload(), Buffer.buffer(originalPayload).getBytes()); } }), anyString(), any(Handler.class)); }
@Test public void testListenerEnqueueWithDefaultQueueingStrategyBecauseOfInvalidConfiguration(TestContext context) throws InterruptedException { setListenerStorageEntryAndTriggerUpdate(buildListenerConfig(new JsonObject().put("type", "discardPayloadXXX"), "x99")); Thread.sleep(1000); String uri = "/playground/server/tests/hooktest/abc123"; String originalPayload = "{\"key\":123}"; PUTRequest putRequest = new PUTRequest(uri, originalPayload); putRequest.addHeader(CONTENT_LENGTH.getName(), "99"); hookHandler.handle(putRequest); Mockito.verify(requestQueue, Mockito.timeout(2000).times(1)).enqueue(Mockito.argThat(new ArgumentMatcher<HttpRequest>() { @Override public boolean matches(Object argument) { HttpRequest req = (HttpRequest) argument; return HttpMethod.PUT == req.getMethod() && req.getUri().contains(uri) && new Integer(99).equals(getInteger(req.getHeaders(), CONTENT_LENGTH)) && Arrays.equals(req.getPayload(), Buffer.buffer(originalPayload).getBytes()); } }), anyString(), any(Handler.class)); }
@Test public void testListenerEnqueueWithDiscardPayloadQueueingStrategy(TestContext context) throws InterruptedException { setListenerStorageEntryAndTriggerUpdate(buildListenerConfig(new JsonObject().put("type", "discardPayload"), "x99")); Thread.sleep(1000); String uri = "/playground/server/tests/hooktest/abc123"; String originalPayload = "{\"key\":123}"; PUTRequest putRequest = new PUTRequest(uri, originalPayload); putRequest.addHeader(CONTENT_LENGTH.getName(), "99"); hookHandler.handle(putRequest); Mockito.verify(requestQueue, Mockito.timeout(2000).times(1)).enqueue(Mockito.argThat(new ArgumentMatcher<HttpRequest>() { @Override public boolean matches(Object argument) { HttpRequest req = (HttpRequest) argument; return HttpMethod.PUT == req.getMethod() && req.getUri().contains(uri) && new Integer(0).equals(getInteger(req.getHeaders(), CONTENT_LENGTH)) && Arrays.equals(req.getPayload(), new byte[0]); } }), anyString(), any(Handler.class)); PUTRequest putRequestWithoutContentLengthHeader = new PUTRequest(uri, originalPayload); hookHandler.handle(putRequestWithoutContentLengthHeader); Mockito.verify(requestQueue, Mockito.timeout(2000).times(1)).enqueue(Mockito.argThat(new ArgumentMatcher<HttpRequest>() { @Override public boolean matches(Object argument) { HttpRequest req = (HttpRequest) argument; return HttpMethod.PUT == req.getMethod() && req.getUri().contains(uri) && !containsHeader(req.getHeaders(), CONTENT_LENGTH) && Arrays.equals(req.getPayload(), new byte[0]); } }), anyString(), any(Handler.class)); }
@Test public void testListenerEnqueueWithReducedPropagationQueueingStrategy(TestContext context) throws InterruptedException { String deviceId = "x99"; long interval = 22; String queue = "listener-hook-http+push+"+deviceId+"+playground+server+tests+hooktest"; setListenerStorageEntryAndTriggerUpdate(buildListenerConfig(new JsonObject().put("type", "reducedPropagation").put("intervalMs", interval), deviceId)); Thread.sleep(1500); String uri = "/playground/server/tests/hooktest/abc123"; String originalPayload = "{\"key\":123}"; PUTRequest putRequest = new PUTRequest(uri, originalPayload); putRequest.addHeader(CONTENT_LENGTH.getName(), "99"); hookHandler.handle(putRequest); String targetUri = "/playground/server/push/v1/devices/"+deviceId+"/playground/server/tests/hooktest/abc123"; Mockito.verify(reducedPropagationManager, Mockito.timeout(2000).times(1)) .processIncomingRequest(eq(HttpMethod.PUT), eq(targetUri), any(MultiMap.class), eq(Buffer.buffer(originalPayload)), eq(queue), eq(interval), any(Handler.class)); }
@Test public void testListenerEnqueueWithInvalidReducedPropagationQueueingStrategy(TestContext context) throws InterruptedException { setListenerStorageEntryAndTriggerUpdate(buildListenerConfig(new JsonObject().put("type", "reducedPropagation").put("intervalMs", "not_a_number"), "x99")); Thread.sleep(1000); String uri = "/playground/server/tests/hooktest/abc123"; String originalPayload = "{\"key\":123}"; PUTRequest putRequest = new PUTRequest(uri, originalPayload); putRequest.addHeader(CONTENT_LENGTH.getName(), "99"); hookHandler.handle(putRequest); Mockito.verify(requestQueue, Mockito.timeout(2000).times(1)).enqueue(Mockito.argThat(new ArgumentMatcher<HttpRequest>() { @Override public boolean matches(Object argument) { HttpRequest req = (HttpRequest) argument; return HttpMethod.PUT == req.getMethod() && req.getUri().contains(uri) && new Integer(99).equals(getInteger(req.getHeaders(), CONTENT_LENGTH)) && Arrays.equals(req.getPayload(), Buffer.buffer(originalPayload).getBytes()); } }), anyString(), any(Handler.class)); }
@Test public void hookRegistration_usesDefaultExpiryIfExpireAfterHeaderIsNegativeNumber(TestContext testContext) { final int[] statusCodePtr = new int[]{ 0 }; final String[] statusMessagePtr = new String[]{ null }; final HttpServerRequest request; { final MultiMap requestHeaders = new CaseInsensitiveHeaders(); requestHeaders.set(ExpiryCheckHandler.EXPIRE_AFTER_HEADER, "-42"); final Buffer requestBody = createMinimalHookBodyAsBuffer(); request = createSimpleRequest(HttpMethod.PUT, "/gateleen/example/_hooks/listeners/http/my-service/my-hook", requestHeaders, requestBody, statusCodePtr, statusMessagePtr ); } hookHandler.handle(request); testContext.assertEquals(200, statusCodePtr[0]); { final String storedHook = storage.getMockData().get(HOOK_ROOT_URI + "registrations/listeners/http+my-service+my-hook+gateleen+example"); testContext.assertNotNull(storedHook); final String expirationTime = new JsonObject(storedHook).getString("expirationTime"); testContext.assertNotNull(expirationTime); testContext.assertEquals("____-__-__T__:__:__.___+__:__".length(), expirationTime.length()); } }
@Test public void hookRegistration_usesDefaultExpiryWhenHeaderContainsCorruptValue(TestContext testContext) { final int[] statusCodePtr = new int[]{ 0 }; final String[] statusMessagePtr = new String[]{ null }; final HttpServerRequest request; { final MultiMap requestHeaders = new CaseInsensitiveHeaders(); requestHeaders.set(ExpiryCheckHandler.EXPIRE_AFTER_HEADER, "This is definitively not a number :)"); final Buffer requestBody = createMinimalHookBodyAsBuffer(); request = createSimpleRequest(HttpMethod.PUT, "/gateleen/example/_hooks/listeners/http/my-service/my-fancy-hook", requestHeaders, requestBody, statusCodePtr, statusMessagePtr ); } hookHandler.handle(request); testContext.assertEquals(200, statusCodePtr[0]); { final String storedHook = storage.getMockData().get(HOOK_ROOT_URI + "registrations/listeners/http+my-service+my-fancy-hook+gateleen+example"); testContext.assertNotNull(storedHook); final String expirationTime = new JsonObject(storedHook).getString("expirationTime"); testContext.assertNotNull(expirationTime); testContext.assertEquals("____-__-__T__:__:__.___+__:__".length(), expirationTime.length()); } }
@Test public void hookRegistration_usesDefaultExpiryIfHeaderIsMissing(TestContext testContext) { final int[] statusCodePtr = new int[]{ 0 }; final String[] statusMessagePtr = new String[]{ null }; final HttpServerRequest request; { final MultiMap requestHeaders = new CaseInsensitiveHeaders(); final Buffer requestBody = createMinimalHookBodyAsBuffer(); request = createSimpleRequest(HttpMethod.PUT, "/gateleen/example/_hooks/listeners/http/my-service/yet-another-hook", requestHeaders, requestBody, statusCodePtr, statusMessagePtr ); } hookHandler.handle(request); testContext.assertEquals(200, statusCodePtr[0]); { final String storedHook = storage.getMockData().get(HOOK_ROOT_URI + "registrations/listeners/http+my-service+yet-another-hook+gateleen+example"); testContext.assertNotNull(storedHook); final String expirationTime = new JsonObject(storedHook).getString("expirationTime"); testContext.assertNotNull(expirationTime); testContext.assertEquals("____-__-__T__:__:__.___+__:__".length(), expirationTime.length()); } }
@Test public void hookRegistration_usesMinusOneIfExpireAfterIsSetToMinusOne(TestContext testContext) { final int[] statusCodePtr = new int[]{ 0 }; final String[] statusMessagePtr = new String[]{ null }; final HttpServerRequest request; { final MultiMap requestHeaders = new CaseInsensitiveHeaders(); requestHeaders.set(ExpiryCheckHandler.EXPIRE_AFTER_HEADER, "-1"); final Buffer requestBody = createMinimalHookBodyAsBuffer(); request = createSimpleRequest(HttpMethod.PUT, "/gateleen/example/_hooks/listeners/http/my-service/and-one-more-again-hook", requestHeaders, requestBody, statusCodePtr, statusMessagePtr ); } hookHandler.handle(request); testContext.assertEquals(200, statusCodePtr[0]); { final String storedHook = storage.getMockData().get(HOOK_ROOT_URI + "registrations/listeners/http+my-service+and-one-more-again-hook+gateleen+example"); testContext.assertNotNull(storedHook); final String expirationTime = new JsonObject(storedHook).getString("expirationTime"); logger.debug("expirationTime is '" + expirationTime + "'"); testContext.assertNull(expirationTime); } }
@Test public void listenerRegistration_acceptOnlyWhitelistedHttpMethods(TestContext testContext) { final int[] statusCodePtr = new int[]{0}; final String[] statusMessagePtr = new String[]{null}; final HttpServerRequest request; { final Buffer requestBody = toBuffer(new JsonObject("{" + " \"methods\": [ \"OPTIONS\" , \"HEAD\" , \"GET\" , \"POST\" , \"PUT\" , \"DELETE\" , \"PATCH\" ]," + " \"destination\": \"/an/example/destination/\"" + "}" )); request = createSimpleRequest(PUT, "/gateleen/example/_hooks/listeners/http/my-service/a-random-hook-8518ul4st8d6944r6k", requestBody, statusCodePtr, statusMessagePtr ); } hookHandler.handle(request); { testContext.assertEquals(200, statusCodePtr[0]); } }
@Test public void listenerRegistration_rejectNotWhitelistedHttpMethods(TestContext testContext) { final List<String> badMethods = Collections.unmodifiableList(new ArrayList<String>() {{ add("CONNECT"); add("TRACE"); add("COPY"); add("LINK"); add("UNLINK"); add("PURGE"); add("LOCK"); add("UNLOCK"); add("PROPFIND"); add("VIEW"); add("FOO"); add("BAR"); add("ASDF"); add("ASSRGHAWERTH"); }}); for (String method : badMethods) { final int[] statusCodePtr = new int[]{0}; final String[] statusMessagePtr = new String[]{null}; final HttpServerRequest request; { final Buffer requestBody = toBuffer(new JsonObject("{" + " \"methods\": [ \"" + method + "\" ]," + " \"destination\": \"/an/example/destination/\"" + "}" )); request = createSimpleRequest(PUT, "/gateleen/example/_hooks/listeners/http/my-service/a-random-hook-8518ul4st8d6944r6k", requestBody, statusCodePtr, statusMessagePtr ); } hookHandler.handle(request); { testContext.assertEquals(400, statusCodePtr[0]); } } }