SearchSource { private Source<JsonObject, NotUsed> retrieveThing(final String thingId, @Nullable final JsonFieldSelector selector) { final RetrieveThing retrieveThing = RetrieveThing.getBuilder(ThingId.of(thingId), getDittoHeaders()) .withSelectedFields(selector) .build(); final CompletionStage<Object> responseFuture = Patterns.ask(conciergeForwarder, retrieveThing, thingsAskTimeout); return Source.fromCompletionStage(responseFuture) .via(expectMsgClass(RetrieveThingResponse.class)) .map(response -> response.getEntity().asObject()); } SearchSource(final ActorRef pubSubMediator,
            final ActorRef conciergeForwarder,
            final Duration thingsAskTimeout,
            final Duration searchAskTimeout,
            @Nullable final JsonFieldSelector fields,
            final JsonFieldSelector sortFields,
            final StreamThings streamThings,
            final String lastThingId); static SearchSourceBuilder newBuilder(); Source<JsonObject, NotUsed> start(final Consumer<ResumeSourceBuilder<?, ?>> configurer); Source<Pair<String, JsonObject>, NotUsed> startAsPair(final Consumer<ResumeSourceBuilder<?, ?>> configurer);  }
@Test public void resumeAtStart() { final JsonArray sortValues = JsonArray.of(997, "t:3"); startTestSearchSource(null, sortValues); sinkProbe.request(200L); conciergeForwarderProbe.expectMsg(streamThings(sortValues)); conciergeForwarderProbe.reply(materializeSourceProbe()); sourceProbe.expectRequest(); sourceProbe.sendNext("t:2").sendComplete(); conciergeForwarderProbe.expectMsg(retrieveThing("t:2", null)); conciergeForwarderProbe.reply(retrieveThingResponse(2)); sinkProbe.expectNext(getThing(2).toJson()) .expectComplete(); }
@Test public void cursorDeleted() { actorSystem.eventStream().setLogLevel(Attributes.logLevelOff()); startTestSearchSource(null, null); sinkProbe.request(200L); conciergeForwarderProbe.expectMsg(streamThings(null)); conciergeForwarderProbe.reply(materializeSourceProbe()); sourceProbe.expectRequest(); sourceProbe.sendNext("t:3"); conciergeForwarderProbe.expectMsg(retrieveThing("t:3", null)); conciergeForwarderProbe.reply(retrieveThingResponse(3)); sinkProbe.expectNext(getThing(3).toJson()); sourceProbe.sendError(new IllegalStateException("mock cursor-not-found error")); conciergeForwarderProbe.expectMsg(retrieveThing("t:3", SORT_FIELDS)); conciergeForwarderProbe.reply(retrieveThingResponse(3)); conciergeForwarderProbe.expectMsg(streamThings(JsonArray.of(997, "t:3"))); conciergeForwarderProbe.reply(materializeSourceProbe()); sourceProbe.expectRequest(); sourceProbe.sendNext("t:2").sendComplete(); conciergeForwarderProbe.expectMsg(retrieveThing("t:2", null)); conciergeForwarderProbe.reply(retrieveThingResponse(2)); sinkProbe.expectNext(getThing(2).toJson()) .expectComplete(); }