ThingsSseRouteBuilder extends RouteDirectives implements SseRouteBuilder { @SuppressWarnings("squid:S1172") @Override public Route build(final RequestContext ctx, final Supplier<CompletionStage<DittoHeaders>> dittoHeadersSupplier) { return concat( rawPathPrefix(PathMatchers.slash().concat(PATH_THINGS), () -> pathEndOrSingleSlash(() -> get(() -> headerValuePF(ACCEPT_HEADER_EXTRACTOR, accept -> { final CompletionStage<DittoHeaders> dittoHeadersCompletionStage = dittoHeadersSupplier.get() .thenApply(ThingsSseRouteBuilder::getDittoHeadersWithCorrelationId); return buildThingsSseRoute(ctx, dittoHeadersCompletionStage); })))), buildSearchSseRoute(ctx, dittoHeadersSupplier) ); } private  ThingsSseRouteBuilder(final ActorRef streamingActor,
            final StreamingConfig streamingConfig,
            final QueryFilterCriteriaFactory queryFilterCriteriaFactory,
            final ActorRef pubSubMediator); static ThingsSseRouteBuilder getInstance(final ActorRef streamingActor,
            final StreamingConfig streamingConfig,
            final ActorRef pubSubMediator); @Override SseRouteBuilder withAuthorizationEnforcer(final SseAuthorizationEnforcer enforcer); @Override ThingsSseRouteBuilder withEventSniffer(final EventSniffer<ServerSentEvent> eventSniffer); @Override SseRouteBuilder withSseConnectionSupervisor(final SseConnectionSupervisor sseConnectionSupervisor); @Override SseRouteBuilder withSignalEnrichmentProvider(@Nullable final GatewaySignalEnrichmentProvider provider); @Override SseRouteBuilder withProxyActor(@Nullable final ActorRef proxyActor); @SuppressWarnings("squid:S1172") // allow unused ctx-Param in order to have a consistent route-"interface" @Override Route build(final RequestContext ctx, final Supplier<CompletionStage<DittoHeaders>> dittoHeadersSupplier);  }
@Test public void getWithAcceptHeaderAndNoQueryParametersOpensSseConnection() { executeThingsRouteTest(HttpRequest.GET(THINGS_ROUTE).addHeader(acceptHeader), StartStreaming.getBuilder(StreamingType.EVENTS, connectionCorrelationId, AuthorizationModelFactory.newAuthContext(DittoAuthorizationContextType.UNSPECIFIED, Collections.emptySet())).build()); }
@Test public void searchWithResumption() { final String lastEventId = "my:lastEventId"; final HttpHeader lastEventIdHeader = HttpHeader.parse("Last-Event-ID", lastEventId); final TestRouteResult routeResult = underTest.run( HttpRequest.GET(SEARCH_ROUTE) .addHeader(acceptHeader) .addHeader(lastEventIdHeader) ); final CompletableFuture<Void> assertions = CompletableFuture.runAsync(() -> { routeResult.assertMediaType(MediaTypes.TEXT_EVENT_STREAM); routeResult.assertStatusCode(StatusCodes.OK); }); final RetrieveThing retrieveThing = proxyActor.expectMsgClass(RetrieveThing.class); final Thing thing = Thing.newBuilder().setId(ThingId.of(lastEventId)).build(); proxyActor.reply(RetrieveThingResponse.of(ThingId.of(lastEventId), thing, retrieveThing.getDittoHeaders() )); final StreamThings streamThings = proxyActor.expectMsgClass(StreamThings.class); replySourceRef(proxyActor, Source.lazily(Source::empty)); assertions.join(); assertThat(streamThings.getSortValues()).contains(JsonArray.of(JsonValue.of(lastEventId))); }
@Test public void getWithAcceptHeaderAndFilterParameterOpensSseConnection() { final String filter = "eq(attributes/manufacturer,\"ACME\")"; final String requestUrl = THINGS_ROUTE + "?filter=" + filter; executeThingsRouteTest(HttpRequest.GET(requestUrl).addHeader(acceptHeader), StartStreaming.getBuilder(StreamingType.EVENTS, connectionCorrelationId, AuthorizationModelFactory.newAuthContext(DittoAuthorizationContextType.UNSPECIFIED, Collections.emptySet())) .withFilter(filter) .build()); }
@Test public void getWithAcceptHeaderAndNamespacesParameterOpensSseConnection() { final Collection<String> namespaces = Lists.list("john", "frum", "tanna"); final String requestUrl = THINGS_ROUTE + "?namespaces=" + String.join(",", namespaces); executeThingsRouteTest(HttpRequest.GET(requestUrl).addHeader(acceptHeader), StartStreaming.getBuilder(StreamingType.EVENTS, connectionCorrelationId, AuthorizationModelFactory.newAuthContext(DittoAuthorizationContextType.UNSPECIFIED, Collections.emptySet())) .withNamespaces(namespaces) .build()); }
@Test public void getWithAcceptHeaderAndExtraFieldsParameterOpensSseConnection() { final JsonFieldSelector extraFields = JsonFieldSelector.newInstance("attributes", "features/location"); final String requestUrl = THINGS_ROUTE + "?extraFields=" + extraFields; executeThingsRouteTest(HttpRequest.GET(requestUrl).addHeader(acceptHeader), StartStreaming.getBuilder(StreamingType.EVENTS, connectionCorrelationId, AuthorizationModelFactory.newAuthContext(DittoAuthorizationContextType.UNSPECIFIED, Collections.emptySet())) .withExtraFields(extraFields) .build()); }