SubscriptionManager extends AbstractActor { private void createSubscription(final CreateSubscription createSubscription) { log.withCorrelationId(createSubscription).info("Processing <{}>", createSubscription); final String subscriptionId = nextSubscriptionId(createSubscription); final Props props = SubscriptionActor.props(idleTimeout, getSender(), createSubscription.getDittoHeaders()); final ActorRef subscriptionActor = getContext().actorOf(props, subscriptionId); final Source<JsonArray, NotUsed> pageSource = getPageSource(createSubscription); connect(subscriptionActor, pageSource); } SubscriptionManager(final Duration idleTimeout,
            final ActorRef pubSubMediator,
            final ActorRef proxyActor,
            final Materializer materializer); static Props props(final Duration idleTimeout,
            final ActorRef pubSubMediator,
            final ActorRef proxyActor,
            final Materializer materializer); @Override Receive createReceive(); static final String ACTOR_NAME; }
@Test public void illegalDemand() { new TestKit(actorSystem) {{ final ActorRef underTest = createSubscriptionManager(); underTest.tell(createSubscription(1), getRef()); final String sid = expectMsgClass(SubscriptionCreated.class).getSubscriptionId(); underTest.tell(RequestFromSubscription.of(sid, 0L, DittoHeaders.empty()), getRef()); expectMsgClass(SubscriptionFailed.class); }}; }
@Test public void parallelSessions() { actorSystem.eventStream().setLogLevel(Attributes.logLevelOff()); final ActorRef underTest = createSubscriptionManager(); final TestProbe probe1 = TestProbe.apply("test1", actorSystem); final TestProbe probe2 = TestProbe.apply("test2", actorSystem); final TestProbe probe3 = TestProbe.apply("test3", actorSystem); final TestProbe probe4 = TestProbe.apply("test4", actorSystem); underTest.tell(createSubscription(1), probe1.ref()); final String prefix = "ABCDEFGHIJKLMNOPQRSTUVWXYZ"; underTest.tell(createSubscription(2).setPrefix(prefix), probe2.ref()); underTest.tell(createSubscription(3), probe3.ref()); underTest.tell(createSubscription(4), probe4.ref()); final String sid1 = probe1.expectMsgClass(SubscriptionCreated.class).getSubscriptionId(); final String sid2 = probe2.expectMsgClass(SubscriptionCreated.class).getSubscriptionId(); final String sid3 = probe3.expectMsgClass(SubscriptionCreated.class).getSubscriptionId(); final String sid4 = probe4.expectMsgClass(SubscriptionCreated.class).getSubscriptionId(); final List<?> sources = List.of( Source.single("t:1"), Source.single("t:2"), Source.from(List.of("t:3", "t:4")), new AskTimeoutException("mock error") ); underTest.tell(request(sid1), probe1.ref()); underTest.tell(request(sid2), probe2.ref()); underTest.tell(request(sid3), probe3.ref()); underTest.tell(request(sid4), probe4.ref()); for (int i = 0; i < 4; ++i) { final StreamThings streamThings = conciergeForwarderProbe.expectMsgClass(StreamThings.class); final ActorRef sender = conciergeForwarderProbe.sender(); final Object source = sources.get(getTag(streamThings) - 1); if (source instanceof Source) { final SourceRef<?> sourceRef = ((Source<?, ?>) source).runWith(StreamRefs.sourceRef(), materializer); sender.tell(sourceRef, ActorRef.noSender()); } else { sender.tell(source, ActorRef.noSender()); } } probe1.expectMsg(hasNext(sid1, "t:1")); probe1.expectMsg(subscriptionComplete(sid1)); probe2.expectMsg(hasNext(sid2, "t:2")); probe2.expectMsg(subscriptionComplete(sid2)); probe3.expectMsg(hasNext(sid3, "t:3")); underTest.tell(CancelSubscription.of(sid3, DittoHeaders.empty()), probe3.ref()); assertThat(probe4.expectMsgClass(SubscriptionFailed.class).getError()) .isInstanceOf(GatewayInternalErrorException.class); CompletableFuture.allOf( CompletableFuture.runAsync( () -> probe3.expectNoMessage(FiniteDuration.apply(250L, TimeUnit.MILLISECONDS))), CompletableFuture.runAsync( () -> probe4.expectNoMessage(FiniteDuration.apply(250L, TimeUnit.MILLISECONDS))) ).join(); }