@Test public void onQueryDisconnected() { AtomicInteger dispatchedSubscriptions = new AtomicInteger(); SubscriptionEvents.SubscribeQuery subscribeQuery = new SubscriptionEvents.SubscribeQuery("Demo", "clientStreamId", QuerySubscription.newBuilder().setClientId("clientId") .setQuery("test").build(), new QueryHandler<QueryProviderInbound>( new FakeStreamObserver<>(), new ClientStreamIdentification("Demo", "clientStreamId"), "component", "client") { @Override public void dispatch(SubscriptionQueryRequest query) { dispatchedSubscriptions.incrementAndGet(); } }); testSubject.on(subscribeQuery); assertEquals(1, dispatchedSubscriptions.get()); testSubject.on(new TopologyEvents.QueryHandlerDisconnected("Demo", "clientId", "clientStreamId")); testSubject.on(subscribeQuery); assertEquals(2, dispatchedSubscriptions.get()); }
@EventListener public void on(ProxiedSubscriptionQueryRequest event) { SubscriptionQueryRequest request = event.subscriptionQueryRequest(); SubscriptionQuery query = event.subscriptionQuery(); QueryHandler handler = registrationCache.find(event.context(), query.getQueryRequest(), event.targetClient()); handler.dispatch(request); }
SubscriptionQueryDispatcher { @EventListener public void on(ProxiedSubscriptionQueryRequest event) { SubscriptionQueryRequest request = event.subscriptionQueryRequest(); SubscriptionQuery query = event.subscriptionQuery(); QueryHandler handler = registrationCache.find(event.context(), query.getQueryRequest(), event.targetClient()); handler.dispatch(request); } }
SubscriptionQueryDispatcher { @EventListener public void on(ProxiedSubscriptionQueryRequest event) { SubscriptionQueryRequest request = event.subscriptionQueryRequest(); SubscriptionQuery query = event.subscriptionQuery(); QueryHandler handler = registrationCache.find(event.context(), query.getQueryRequest(), event.targetClient()); handler.dispatch(request); } SubscriptionQueryDispatcher(Iterable<ContextSubscriptionQuery> directSubscriptions,
                                       QueryRegistrationCache registrationCache); }
SubscriptionQueryDispatcher { @EventListener public void on(ProxiedSubscriptionQueryRequest event) { SubscriptionQueryRequest request = event.subscriptionQueryRequest(); SubscriptionQuery query = event.subscriptionQuery(); QueryHandler handler = registrationCache.find(event.context(), query.getQueryRequest(), event.targetClient()); handler.dispatch(request); } SubscriptionQueryDispatcher(Iterable<ContextSubscriptionQuery> directSubscriptions,
                                       QueryRegistrationCache registrationCache); @EventListener void on(ProxiedSubscriptionQueryRequest event); @EventListener void on(SubscriptionQueryRequestEvent event); @EventListener void on(SubscriptionQueryCanceled evt); @EventListener void on(SubscriptionEvents.SubscribeQuery event); @EventListener void on(QueryHandlerDisconnected event); }
SubscriptionQueryDispatcher { @EventListener public void on(ProxiedSubscriptionQueryRequest event) { SubscriptionQueryRequest request = event.subscriptionQueryRequest(); SubscriptionQuery query = event.subscriptionQuery(); QueryHandler handler = registrationCache.find(event.context(), query.getQueryRequest(), event.targetClient()); handler.dispatch(request); } SubscriptionQueryDispatcher(Iterable<ContextSubscriptionQuery> directSubscriptions,
                                       QueryRegistrationCache registrationCache); @EventListener void on(ProxiedSubscriptionQueryRequest event); @EventListener void on(SubscriptionQueryRequestEvent event); @EventListener void on(SubscriptionQueryCanceled evt); @EventListener void on(SubscriptionEvents.SubscribeQuery event); @EventListener void on(QueryHandlerDisconnected event);  }