#11 Subscription Query - Work In Progress

AxonFramework · Jun 5, 2018 · 05a6ce0 · 05a6ce0
1 parent bfe2588
commit 05a6ce0
Show file tree

Hide file tree

Showing 16 changed files with 515 additions and 779 deletions.
diff --git a/src/main/java/io/axoniq/axonhub/client/query/AxonHubQueryBus.java b/src/main/java/io/axoniq/axonhub/client/query/AxonHubQueryBus.java
@@ -139,8 +139,7 @@ public <R> Registration subscribe(String queryName, Type responseType,
     public <Q, R> CompletableFuture<QueryResponseMessage<R>> query(QueryMessage<Q, R> queryMessage) {
         QueryMessage<Q, R> interceptedQuery = dispatchInterceptors.intercept(queryMessage);
         CompletableFuture<QueryResponseMessage<R>> completableFuture = new CompletableFuture<>();
-        QueryServiceGrpc.newStub(platformConnectionManager.getChannel())
+        queryServiceStub()
-                        .withInterceptors(interceptors)
                         .query(serializer.serializeRequest(interceptedQuery, 1,
                                                            TimeUnit.HOURS.toMillis(1), priorityCalculator.determinePriority(interceptedQuery)),
                                new StreamObserver<QueryResponse>() {
@@ -166,12 +165,15 @@ public void onCompleted() {
         return completableFuture;
     }
 
+    public QueryServiceGrpc.QueryServiceStub queryServiceStub() {
+        return QueryServiceGrpc.newStub(platformConnectionManager.getChannel()).withInterceptors(interceptors);
+    }
+
     @Override
     public <Q, R> Stream<QueryResponseMessage<R>> scatterGather(QueryMessage<Q, R> queryMessage, long timeout, TimeUnit timeUnit) {
         QueryMessage<Q, R> interceptedQuery = dispatchInterceptors.intercept(queryMessage);
         QueueBackedSpliterator<QueryResponseMessage<R>> resultSpliterator = new QueueBackedSpliterator<>(timeout, timeUnit);
-        QueryServiceGrpc.newStub(platformConnectionManager.getChannel())
+        queryServiceStub()
-                        .withInterceptors(interceptors)
                         .withDeadlineAfter(timeout, timeUnit)
                         .query(serializer.serializeRequest(interceptedQuery, -1, timeUnit.toMillis(timeout),
                                                            priorityCalculator.determinePriority(interceptedQuery)),
@@ -400,7 +402,7 @@ public int hashCode() {
         }
     }
 
-    void publish(QueryProviderOutbound providerOutbound){
+    public void publish(QueryProviderOutbound providerOutbound){
         this.queryProvider.getSubscriberObserver().onNext(providerOutbound);
     }
 

diff --git a/src/main/java/io/axoniq/axonhub/client/query/EnhancedAxonHubQueryBus.java b/src/main/java/io/axoniq/axonhub/client/query/EnhancedAxonHubQueryBus.java
diff --git a/src/main/java/io/axoniq/axonhub/client/query/subscription/AxonHubUpdateDispatcher.java b/src/main/java/io/axoniq/axonhub/client/query/subscription/AxonHubUpdateDispatcher.java
diff --git a/src/main/java/io/axoniq/axonhub/client/query/subscription/AxonHubUpdateEmitter.java b/src/main/java/io/axoniq/axonhub/client/query/subscription/AxonHubUpdateEmitter.java