@Test void testSubscribe() { testSubject.subscribe("test", String.class, Message::getPayload); assertEquals(1, testSubject.getSubscriptions().size()); assertEquals(1, testSubject.getSubscriptions().values().iterator().next().size()); testSubject.subscribe("test", String.class, (q) -> "aa" + q.getPayload()); assertEquals(1, testSubject.getSubscriptions().size()); assertEquals(2, testSubject.getSubscriptions().values().iterator().next().size()); testSubject.subscribe("test2", String.class, (q) -> "aa" + q.getPayload()); assertEquals(2, testSubject.getSubscriptions().size()); }
@Override public <R> Registration subscribe(String queryName, Type responseType, MessageHandler<? super QueryMessage<?, R>> handler) { CopyOnWriteArrayList<QuerySubscription> handlers = subscriptions.computeIfAbsent(queryName, k -> new CopyOnWriteArrayList<>()); QuerySubscription<R> querySubscription = new QuerySubscription<>(responseType, handler); handlers.addIfAbsent(querySubscription); return () -> unsubscribe(queryName, querySubscription); }
SimpleQueryBus implements QueryBus { @Override public <R> Registration subscribe(String queryName, Type responseType, MessageHandler<? super QueryMessage<?, R>> handler) { CopyOnWriteArrayList<QuerySubscription> handlers = subscriptions.computeIfAbsent(queryName, k -> new CopyOnWriteArrayList<>()); QuerySubscription<R> querySubscription = new QuerySubscription<>(responseType, handler); handlers.addIfAbsent(querySubscription); return () -> unsubscribe(queryName, querySubscription); } }
SimpleQueryBus implements QueryBus { @Override public <R> Registration subscribe(String queryName, Type responseType, MessageHandler<? super QueryMessage<?, R>> handler) { CopyOnWriteArrayList<QuerySubscription> handlers = subscriptions.computeIfAbsent(queryName, k -> new CopyOnWriteArrayList<>()); QuerySubscription<R> querySubscription = new QuerySubscription<>(responseType, handler); handlers.addIfAbsent(querySubscription); return () -> unsubscribe(queryName, querySubscription); } protected  SimpleQueryBus(Builder builder); }
SimpleQueryBus implements QueryBus { @Override public <R> Registration subscribe(String queryName, Type responseType, MessageHandler<? super QueryMessage<?, R>> handler) { CopyOnWriteArrayList<QuerySubscription> handlers = subscriptions.computeIfAbsent(queryName, k -> new CopyOnWriteArrayList<>()); QuerySubscription<R> querySubscription = new QuerySubscription<>(responseType, handler); handlers.addIfAbsent(querySubscription); return () -> unsubscribe(queryName, querySubscription); } protected  SimpleQueryBus(Builder builder); static Builder builder(); @Override Registration subscribe(String queryName,
                                      Type responseType,
                                      MessageHandler<? super QueryMessage<?, R>> handler); @Override CompletableFuture<QueryResponseMessage<R>> query(QueryMessage<Q, R> query); @Override Stream<QueryResponseMessage<R>> scatterGather(QueryMessage<Q, R> query, long timeout, TimeUnit unit); @SuppressWarnings("unchecked") @Override SubscriptionQueryResult<QueryResponseMessage<I>, SubscriptionQueryUpdateMessage<U>> subscriptionQuery(
            SubscriptionQueryMessage<Q, I, U> query,
            SubscriptionQueryBackpressure backpressure,
            int updateBufferSize); @Override QueryUpdateEmitter queryUpdateEmitter(); @Override Registration registerHandlerInterceptor(MessageHandlerInterceptor<? super QueryMessage<?, ?>> interceptor); @Override Registration registerDispatchInterceptor(
            MessageDispatchInterceptor<? super QueryMessage<?, ?>> interceptor); }
SimpleQueryBus implements QueryBus { @Override public <R> Registration subscribe(String queryName, Type responseType, MessageHandler<? super QueryMessage<?, R>> handler) { CopyOnWriteArrayList<QuerySubscription> handlers = subscriptions.computeIfAbsent(queryName, k -> new CopyOnWriteArrayList<>()); QuerySubscription<R> querySubscription = new QuerySubscription<>(responseType, handler); handlers.addIfAbsent(querySubscription); return () -> unsubscribe(queryName, querySubscription); } protected  SimpleQueryBus(Builder builder); static Builder builder(); @Override Registration subscribe(String queryName,
                                      Type responseType,
                                      MessageHandler<? super QueryMessage<?, R>> handler); @Override CompletableFuture<QueryResponseMessage<R>> query(QueryMessage<Q, R> query); @Override Stream<QueryResponseMessage<R>> scatterGather(QueryMessage<Q, R> query, long timeout, TimeUnit unit); @SuppressWarnings("unchecked") @Override SubscriptionQueryResult<QueryResponseMessage<I>, SubscriptionQueryUpdateMessage<U>> subscriptionQuery(
            SubscriptionQueryMessage<Q, I, U> query,
            SubscriptionQueryBackpressure backpressure,
            int updateBufferSize); @Override QueryUpdateEmitter queryUpdateEmitter(); @Override Registration registerHandlerInterceptor(MessageHandlerInterceptor<? super QueryMessage<?, ?>> interceptor); @Override Registration registerDispatchInterceptor(
            MessageDispatchInterceptor<? super QueryMessage<?, ?>> interceptor);  }