test_class.declarator: actorSystem
test_class.modifier: private
test_class.original_string: private ActorSystem actorSystem;
test_class.type: ActorSystem
test_class.var_name: actorSystem
test_class.declarator: materializer
test_class.modifier: private
test_class.original_string: private Materializer materializer;
test_class.type: Materializer
test_class.var_name: materializer
test_class.declarator: pubSubMediatorProbe
test_class.modifier: private
test_class.original_string: private TestProbe pubSubMediatorProbe;
test_class.type: TestProbe
test_class.var_name: pubSubMediatorProbe
test_class.declarator: conciergeForwarderProbe
test_class.modifier: private
test_class.original_string: private TestProbe conciergeForwarderProbe;
test_class.type: TestProbe
test_class.var_name: conciergeForwarderProbe
test_class.file: services/utils/search/src/test/java/org/eclipse/ditto/services/utils/search/SubscriptionManagerTest.java
test_class.identifier: SubscriptionManagerTest
test_case.body: @Test
      public void parallelSessions() {
          // Suppressing logs due to stacktrace for test4. Comment out to see logs.
          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);
  
          // expect success with results
          underTest.tell(createSubscription(1), probe1.ref());
          // expect success with prefix
          final String prefix = "ABCDEFGHIJKLMNOPQRSTUVWXYZ";
          underTest.tell(createSubscription(2).setPrefix(prefix), probe2.ref());
          // expect cancellation
          underTest.tell(createSubscription(3), probe3.ref());
          // expect upstream failure
          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());
  
          // there should be no upstream request until downstream requests.
          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();
      }
test_case.class_method_signature: SubscriptionManagerTest.parallelSessions()
test_case.full_signature: @Test public void parallelSessions()
test_case.identifier: parallelSessions
test_case.invocations: setLogLevel eventStream logLevelOff createSubscriptionManager apply apply apply apply tell createSubscription ref tell setPrefix createSubscription ref tell createSubscription ref tell createSubscription ref getSubscriptionId expectMsgClass getSubscriptionId expectMsgClass getSubscriptionId expectMsgClass getSubscriptionId expectMsgClass of single single from of tell request ref tell request ref tell request ref tell request ref expectMsgClass sender get getTag runWith sourceRef tell noSender tell noSender expectMsg hasNext expectMsg subscriptionComplete expectMsg hasNext expectMsg subscriptionComplete expectMsg hasNext tell of empty ref isInstanceOf assertThat getError expectMsgClass join allOf runAsync expectNoMessage apply runAsync expectNoMessage apply
test_case.modifiers: @Test public
test_case.parameters: ()
test_case.return: void
test_case.signature: void parallelSessions()
focal_class.declarator: ACTOR_NAME = "subscriptionManager"
focal_class.modifier: public static final
focal_class.original_string: public static final String ACTOR_NAME = "subscriptionManager";
focal_class.type: String
focal_class.var_name: ACTOR_NAME
focal_class.declarator: idleTimeout
focal_class.modifier: private final
focal_class.original_string: private final Duration idleTimeout;
focal_class.type: Duration
focal_class.var_name: idleTimeout
focal_class.declarator: pubSubMediator
focal_class.modifier: private final
focal_class.original_string: private final ActorRef pubSubMediator;
focal_class.type: ActorRef
focal_class.var_name: pubSubMediator
focal_class.declarator: proxyActor
focal_class.modifier: private final
focal_class.original_string: private final ActorRef proxyActor;
focal_class.type: ActorRef
focal_class.var_name: proxyActor
focal_class.declarator: materializer
focal_class.modifier: private final
focal_class.original_string: private final Materializer materializer;
focal_class.type: Materializer
focal_class.var_name: materializer
focal_class.declarator: log
focal_class.modifier: private final
focal_class.original_string: private final DittoDiagnosticLoggingAdapter log;
focal_class.type: DittoDiagnosticLoggingAdapter
focal_class.var_name: log
focal_class.declarator: defaultPageSize
focal_class.modifier: private final
focal_class.original_string: private final int defaultPageSize;
focal_class.type: int
focal_class.var_name: defaultPageSize
focal_class.declarator: maxPageSize
focal_class.modifier: private final
focal_class.original_string: private final int maxPageSize;
focal_class.type: int
focal_class.var_name: maxPageSize
focal_class.declarator: subscriptionIdCounter = 0
focal_class.modifier: private
focal_class.original_string: private int subscriptionIdCounter = 0;
focal_class.type: int
focal_class.var_name: subscriptionIdCounter
focal_class.file: services/utils/search/src/main/java/org/eclipse/ditto/services/utils/search/SubscriptionManager.java
focal_class.identifier: SubscriptionManager
focal_class.class_method_signature: SubscriptionManager.SubscriptionManager(final Duration idleTimeout,
              final ActorRef pubSubMediator,
              final ActorRef proxyActor,
              final Materializer materializer)
focal_class.full_signature:   SubscriptionManager(final Duration idleTimeout,
              final ActorRef pubSubMediator,
              final ActorRef proxyActor,
              final Materializer materializer)
focal_class.identifier: SubscriptionManager
focal_class.parameters: (final Duration idleTimeout,
              final ActorRef pubSubMediator,
              final ActorRef proxyActor,
              final Materializer materializer)
focal_class.signature:  SubscriptionManager(final Duration idleTimeout,
              final ActorRef pubSubMediator,
              final ActorRef proxyActor,
              final Materializer materializer)
focal_class.class_method_signature: SubscriptionManager.props(final Duration idleTimeout,
              final ActorRef pubSubMediator,
              final ActorRef proxyActor,
              final Materializer materializer)
focal_class.full_signature: public static Props props(final Duration idleTimeout,
              final ActorRef pubSubMediator,
              final ActorRef proxyActor,
              final Materializer materializer)
focal_class.identifier: props
focal_class.modifiers: public static
focal_class.parameters: (final Duration idleTimeout,
              final ActorRef pubSubMediator,
              final ActorRef proxyActor,
              final Materializer materializer)
focal_class.return: Props
focal_class.signature: Props props(final Duration idleTimeout,
              final ActorRef pubSubMediator,
              final ActorRef proxyActor,
              final Materializer materializer)
focal_class.class_method_signature: SubscriptionManager.asJsonArray(final Collection<String> strings)
focal_class.full_signature: private static JsonArray asJsonArray(final Collection<String> strings)
focal_class.identifier: asJsonArray
focal_class.modifiers: private static
focal_class.parameters: (final Collection<String> strings)
focal_class.return: JsonArray
focal_class.signature: JsonArray asJsonArray(final Collection<String> strings)
focal_class.class_method_signature: SubscriptionManager.getPageSize(@Nullable final String optionString)
focal_class.full_signature: private int getPageSize(@Nullable final String optionString)
focal_class.identifier: getPageSize
focal_class.modifiers: private
focal_class.parameters: (@Nullable final String optionString)
focal_class.return: int
focal_class.signature: int getPageSize(@Nullable final String optionString)
focal_class.class_method_signature: SubscriptionManager.createReceive()
focal_class.full_signature: @Override public Receive createReceive()
focal_class.identifier: createReceive
focal_class.modifiers: @Override public
focal_class.parameters: ()
focal_class.return: Receive
focal_class.signature: Receive createReceive()
focal_class.class_method_signature: SubscriptionManager.requestSubscription(final RequestFromSubscription requestFromSubscription)
focal_class.full_signature: private void requestSubscription(final RequestFromSubscription requestFromSubscription)
focal_class.identifier: requestSubscription
focal_class.modifiers: private
focal_class.parameters: (final RequestFromSubscription requestFromSubscription)
focal_class.return: void
focal_class.signature: void requestSubscription(final RequestFromSubscription requestFromSubscription)
focal_class.class_method_signature: SubscriptionManager.cancelSubscription(final CancelSubscription cancelSubscription)
focal_class.full_signature: private void cancelSubscription(final CancelSubscription cancelSubscription)
focal_class.identifier: cancelSubscription
focal_class.modifiers: private
focal_class.parameters: (final CancelSubscription cancelSubscription)
focal_class.return: void
focal_class.signature: void cancelSubscription(final CancelSubscription cancelSubscription)
focal_class.class_method_signature: SubscriptionManager.forwardToChild(final String subscriptionId, final ThingSearchCommand<?> command)
focal_class.full_signature: private void forwardToChild(final String subscriptionId, final ThingSearchCommand<?> command)
focal_class.identifier: forwardToChild
focal_class.modifiers: private
focal_class.parameters: (final String subscriptionId, final ThingSearchCommand<?> command)
focal_class.return: void
focal_class.signature: void forwardToChild(final String subscriptionId, final ThingSearchCommand<?> command)
focal_class.class_method_signature: SubscriptionManager.createSubscription(final CreateSubscription createSubscription)
focal_class.full_signature: private void createSubscription(final CreateSubscription createSubscription)
focal_class.identifier: createSubscription
focal_class.modifiers: private
focal_class.parameters: (final CreateSubscription createSubscription)
focal_class.return: void
focal_class.signature: void createSubscription(final CreateSubscription createSubscription)
focal_class.class_method_signature: SubscriptionManager.connect(final ActorRef subscriptionActor, final Source<JsonArray, NotUsed> pageSource)
focal_class.full_signature: private void connect(final ActorRef subscriptionActor, final Source<JsonArray, NotUsed> pageSource)
focal_class.identifier: connect
focal_class.modifiers: private
focal_class.parameters: (final ActorRef subscriptionActor, final Source<JsonArray, NotUsed> pageSource)
focal_class.return: void
focal_class.signature: void connect(final ActorRef subscriptionActor, final Source<JsonArray, NotUsed> pageSource)
focal_class.class_method_signature: SubscriptionManager.getPageSource(final CreateSubscription createSubscription)
focal_class.full_signature: private Source<JsonArray, NotUsed> getPageSource(final CreateSubscription createSubscription)
focal_class.identifier: getPageSource
focal_class.modifiers: private
focal_class.parameters: (final CreateSubscription createSubscription)
focal_class.return: Source<JsonArray, NotUsed>
focal_class.signature: Source<JsonArray, NotUsed> getPageSource(final CreateSubscription createSubscription)
focal_class.class_method_signature: SubscriptionManager.nextSubscriptionId(final CreateSubscription createSubscription)
focal_class.full_signature: private String nextSubscriptionId(final CreateSubscription createSubscription)
focal_class.identifier: nextSubscriptionId
focal_class.modifiers: private
focal_class.parameters: (final CreateSubscription createSubscription)
focal_class.return: String
focal_class.signature: String nextSubscriptionId(final CreateSubscription createSubscription)
focal_class.class_method_signature: SubscriptionManager.lazify(final Source<T, ?> upstream)
focal_class.full_signature: private static Source<T, ?> lazify(final Source<T, ?> upstream)
focal_class.identifier: lazify
focal_class.modifiers: private static
focal_class.parameters: (final Source<T, ?> upstream)
focal_class.return: Source<T, ?>
focal_class.signature: Source<T, ?> lazify(final Source<T, ?> upstream)
focal_class.superclass: extends AbstractActor
focal_method.body: 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);
      }
focal_method.class_method_signature: SubscriptionManager.createSubscription(final CreateSubscription createSubscription)
focal_method.full_signature: private void createSubscription(final CreateSubscription createSubscription)
focal_method.identifier: createSubscription
focal_method.invocations: info withCorrelationId nextSubscriptionId props getSender getDittoHeaders actorOf getContext getPageSource connect
focal_method.modifiers: private
focal_method.parameters: (final CreateSubscription createSubscription)
focal_method.return: void
focal_method.signature: void createSubscription(final CreateSubscription createSubscription)