@Test public void find() { queryRegistrationCache.add(new QueryDefinition(Topology.DEFAULT_CONTEXT, "test"), "test", new DirectQueryHandler(dummyStreamObserver, new ClientStreamIdentification(Topology.DEFAULT_CONTEXT, "client"), "component", "client")); QueryRequest request = QueryRequest.newBuilder().setQuery("test").build(); assertNotNull(queryRegistrationCache.find(Topology.DEFAULT_CONTEXT, request,"client")); assertNull(queryRegistrationCache.find(Topology.DEFAULT_CONTEXT, request, "client1")); }
public Set<QueryHandler> find(String context, QueryRequest request) { QueryDefinition queryDefinition = new QueryDefinition(context, request.getQuery()); QueryInformation queryInformation = registrationsPerQuery.get(queryDefinition); if (queryInformation == null) { return Collections.emptySet(); } Set<ClientStreamIdentification> candidates = queryInformation.getHandlersPerComponent() .values() .stream() .flatMap(Collection::stream).collect(toSet()); Set<ClientStreamIdentification> filteredCandidates = metaDataBasedNodeSelector.apply(request.getMetaDataMap(), candidates); return queryInformation.getHandlersPerComponent().entrySet().stream() .map(entry -> pickOne(queryDefinition, entry.getKey(), entry.getValue(), filteredCandidates)) .filter(Objects::nonNull) .collect(toSet()); }
QueryRegistrationCache { public Set<QueryHandler> find(String context, QueryRequest request) { QueryDefinition queryDefinition = new QueryDefinition(context, request.getQuery()); QueryInformation queryInformation = registrationsPerQuery.get(queryDefinition); if (queryInformation == null) { return Collections.emptySet(); } Set<ClientStreamIdentification> candidates = queryInformation.getHandlersPerComponent() .values() .stream() .flatMap(Collection::stream).collect(toSet()); Set<ClientStreamIdentification> filteredCandidates = metaDataBasedNodeSelector.apply(request.getMetaDataMap(), candidates); return queryInformation.getHandlersPerComponent().entrySet().stream() .map(entry -> pickOne(queryDefinition, entry.getKey(), entry.getValue(), filteredCandidates)) .filter(Objects::nonNull) .collect(toSet()); } }
QueryRegistrationCache { public Set<QueryHandler> find(String context, QueryRequest request) { QueryDefinition queryDefinition = new QueryDefinition(context, request.getQuery()); QueryInformation queryInformation = registrationsPerQuery.get(queryDefinition); if (queryInformation == null) { return Collections.emptySet(); } Set<ClientStreamIdentification> candidates = queryInformation.getHandlersPerComponent() .values() .stream() .flatMap(Collection::stream).collect(toSet()); Set<ClientStreamIdentification> filteredCandidates = metaDataBasedNodeSelector.apply(request.getMetaDataMap(), candidates); return queryInformation.getHandlersPerComponent().entrySet().stream() .map(entry -> pickOne(queryDefinition, entry.getKey(), entry.getValue(), filteredCandidates)) .filter(Objects::nonNull) .collect(toSet()); } @Autowired  QueryRegistrationCache(QueryHandlerSelector queryHandlerSelector,
                                  BiFunction<Map<String, MetaDataValue>, Set<ClientStreamIdentification>, Set<ClientStreamIdentification>> metaDataBasedNodeSelector);  QueryRegistrationCache(QueryHandlerSelector queryHandlerSelector); }
QueryRegistrationCache { public Set<QueryHandler> find(String context, QueryRequest request) { QueryDefinition queryDefinition = new QueryDefinition(context, request.getQuery()); QueryInformation queryInformation = registrationsPerQuery.get(queryDefinition); if (queryInformation == null) { return Collections.emptySet(); } Set<ClientStreamIdentification> candidates = queryInformation.getHandlersPerComponent() .values() .stream() .flatMap(Collection::stream).collect(toSet()); Set<ClientStreamIdentification> filteredCandidates = metaDataBasedNodeSelector.apply(request.getMetaDataMap(), candidates); return queryInformation.getHandlersPerComponent().entrySet().stream() .map(entry -> pickOne(queryDefinition, entry.getKey(), entry.getValue(), filteredCandidates)) .filter(Objects::nonNull) .collect(toSet()); } @Autowired  QueryRegistrationCache(QueryHandlerSelector queryHandlerSelector,
                                  BiFunction<Map<String, MetaDataValue>, Set<ClientStreamIdentification>, Set<ClientStreamIdentification>> metaDataBasedNodeSelector);  QueryRegistrationCache(QueryHandlerSelector queryHandlerSelector); @EventListener void on(SubscriptionEvents.UnsubscribeQuery event); @EventListener void on(SubscriptionEvents.SubscribeQuery event); void remove(ClientStreamIdentification clientStreamIdentification); void remove(QueryDefinition queryDefinition, ClientStreamIdentification clientId); void add(QueryDefinition queryDefinition, String resultName,
                    QueryHandler queryHandler); Set<QueryHandler> find(String context, QueryRequest request); Collection<QueryHandler> findAll(String context, QueryRequest request); Map<QueryDefinition, Map<String, Set<QueryHandler<?>>>> getAll(); List<QueryRegistration> getForClient(ClientStreamIdentification client); QueryHandler find(String context, QueryRequest request, String clientStreamId); Set<ClientStreamIdentification> getClients(); Set<String> getResponseTypes(QueryDefinition key); }
QueryRegistrationCache { public Set<QueryHandler> find(String context, QueryRequest request) { QueryDefinition queryDefinition = new QueryDefinition(context, request.getQuery()); QueryInformation queryInformation = registrationsPerQuery.get(queryDefinition); if (queryInformation == null) { return Collections.emptySet(); } Set<ClientStreamIdentification> candidates = queryInformation.getHandlersPerComponent() .values() .stream() .flatMap(Collection::stream).collect(toSet()); Set<ClientStreamIdentification> filteredCandidates = metaDataBasedNodeSelector.apply(request.getMetaDataMap(), candidates); return queryInformation.getHandlersPerComponent().entrySet().stream() .map(entry -> pickOne(queryDefinition, entry.getKey(), entry.getValue(), filteredCandidates)) .filter(Objects::nonNull) .collect(toSet()); } @Autowired  QueryRegistrationCache(QueryHandlerSelector queryHandlerSelector,
                                  BiFunction<Map<String, MetaDataValue>, Set<ClientStreamIdentification>, Set<ClientStreamIdentification>> metaDataBasedNodeSelector);  QueryRegistrationCache(QueryHandlerSelector queryHandlerSelector); @EventListener void on(SubscriptionEvents.UnsubscribeQuery event); @EventListener void on(SubscriptionEvents.SubscribeQuery event); void remove(ClientStreamIdentification clientStreamIdentification); void remove(QueryDefinition queryDefinition, ClientStreamIdentification clientId); void add(QueryDefinition queryDefinition, String resultName,
                    QueryHandler queryHandler); Set<QueryHandler> find(String context, QueryRequest request); Collection<QueryHandler> findAll(String context, QueryRequest request); Map<QueryDefinition, Map<String, Set<QueryHandler<?>>>> getAll(); List<QueryRegistration> getForClient(ClientStreamIdentification client); QueryHandler find(String context, QueryRequest request, String clientStreamId); Set<ClientStreamIdentification> getClients(); Set<String> getResponseTypes(QueryDefinition key);  }