@Test public void queryEvents() { FakeStreamObserver<QueryEventsResponse> responseObserver = new FakeStreamObserver<>(); AtomicReference<StreamObserver<QueryEventsResponse>> eventStoreOutputStreamRef = new AtomicReference<>(); StreamObserver<QueryEventsRequest> eventStoreResponseStream = new StreamObserver<QueryEventsRequest>() { @Override public void onNext(QueryEventsRequest value) { StreamObserver<QueryEventsResponse> responseStream = eventStoreOutputStreamRef.get(); responseStream.onNext(QueryEventsResponse.newBuilder().build()); responseStream.onCompleted(); } @Override public void onError(Throwable t) { } @Override public void onCompleted() { } }; when(eventStoreClient.queryEvents(any(), any(StreamObserver.class))).then(a -> { eventStoreOutputStreamRef.set((StreamObserver<QueryEventsResponse>) a.getArguments()[1]); return eventStoreResponseStream; }); StreamObserver<QueryEventsRequest> inputStream = testSubject.queryEvents(responseObserver); inputStream.onNext(QueryEventsRequest.newBuilder().build()); verify(eventStoreLocator).getEventStore(Topology.DEFAULT_CONTEXT, false); assertEquals(1, responseObserver.values().size()); responseObserver = new FakeStreamObserver<>(); inputStream = testSubject.queryEvents(responseObserver); inputStream.onNext(QueryEventsRequest.newBuilder() .setForceReadFromLeader(true) .build()); verify(eventStoreLocator).getEventStore(Topology.DEFAULT_CONTEXT, true); assertEquals(1, responseObserver.completedCount()); }
public StreamObserver<QueryEventsRequest> queryEvents(StreamObserver<QueryEventsResponse> responseObserver0) { String context = contextProvider.getContext(); ForwardingStreamObserver<QueryEventsResponse> responseObserver = new ForwardingStreamObserver<>(logger, "queryEvents", responseObserver0); return new StreamObserver<QueryEventsRequest>() { private final AtomicReference<StreamObserver<QueryEventsRequest>> requestObserver = new AtomicReference<>(); @Override public void onNext(QueryEventsRequest request) { if (requestObserver.get() == null) { EventStore eventStore = eventStoreLocator.getEventStore(context, request.getForceReadFromLeader()); if (eventStore == null) { responseObserver.onError(new MessagingPlatformException(ErrorCode.NO_EVENTSTORE, NO_EVENT_STORE_CONFIGURED + context)); return; } requestObserver.set(eventStore.queryEvents(context, responseObserver)); } try { requestObserver.get().onNext(request); } catch (Exception reason) { logger.warn("{}: Error forwarding request to event store: {}", context, reason.getMessage()); StreamObserverUtils.complete(requestObserver.get()); } } @Override public void onError(Throwable reason) { if (!ExceptionUtils.isCancelled(reason)) { logger.warn("Error on connection from client: {}", reason.getMessage()); } cleanup(); } @Override public void onCompleted() { cleanup(); StreamObserverUtils.complete(responseObserver); } private void cleanup() { StreamObserverUtils.complete(requestObserver.get()); } }; }
EventDispatcher implements AxonServerClientService { public StreamObserver<QueryEventsRequest> queryEvents(StreamObserver<QueryEventsResponse> responseObserver0) { String context = contextProvider.getContext(); ForwardingStreamObserver<QueryEventsResponse> responseObserver = new ForwardingStreamObserver<>(logger, "queryEvents", responseObserver0); return new StreamObserver<QueryEventsRequest>() { private final AtomicReference<StreamObserver<QueryEventsRequest>> requestObserver = new AtomicReference<>(); @Override public void onNext(QueryEventsRequest request) { if (requestObserver.get() == null) { EventStore eventStore = eventStoreLocator.getEventStore(context, request.getForceReadFromLeader()); if (eventStore == null) { responseObserver.onError(new MessagingPlatformException(ErrorCode.NO_EVENTSTORE, NO_EVENT_STORE_CONFIGURED + context)); return; } requestObserver.set(eventStore.queryEvents(context, responseObserver)); } try { requestObserver.get().onNext(request); } catch (Exception reason) { logger.warn("{}: Error forwarding request to event store: {}", context, reason.getMessage()); StreamObserverUtils.complete(requestObserver.get()); } } @Override public void onError(Throwable reason) { if (!ExceptionUtils.isCancelled(reason)) { logger.warn("Error on connection from client: {}", reason.getMessage()); } cleanup(); } @Override public void onCompleted() { cleanup(); StreamObserverUtils.complete(responseObserver); } private void cleanup() { StreamObserverUtils.complete(requestObserver.get()); } }; } }
EventDispatcher implements AxonServerClientService { public StreamObserver<QueryEventsRequest> queryEvents(StreamObserver<QueryEventsResponse> responseObserver0) { String context = contextProvider.getContext(); ForwardingStreamObserver<QueryEventsResponse> responseObserver = new ForwardingStreamObserver<>(logger, "queryEvents", responseObserver0); return new StreamObserver<QueryEventsRequest>() { private final AtomicReference<StreamObserver<QueryEventsRequest>> requestObserver = new AtomicReference<>(); @Override public void onNext(QueryEventsRequest request) { if (requestObserver.get() == null) { EventStore eventStore = eventStoreLocator.getEventStore(context, request.getForceReadFromLeader()); if (eventStore == null) { responseObserver.onError(new MessagingPlatformException(ErrorCode.NO_EVENTSTORE, NO_EVENT_STORE_CONFIGURED + context)); return; } requestObserver.set(eventStore.queryEvents(context, responseObserver)); } try { requestObserver.get().onNext(request); } catch (Exception reason) { logger.warn("{}: Error forwarding request to event store: {}", context, reason.getMessage()); StreamObserverUtils.complete(requestObserver.get()); } } @Override public void onError(Throwable reason) { if (!ExceptionUtils.isCancelled(reason)) { logger.warn("Error on connection from client: {}", reason.getMessage()); } cleanup(); } @Override public void onCompleted() { cleanup(); StreamObserverUtils.complete(responseObserver); } private void cleanup() { StreamObserverUtils.complete(requestObserver.get()); } }; } EventDispatcher(EventStoreLocator eventStoreLocator,
                           ContextProvider contextProvider,
                           MeterFactory meterFactory); }
EventDispatcher implements AxonServerClientService { public StreamObserver<QueryEventsRequest> queryEvents(StreamObserver<QueryEventsResponse> responseObserver0) { String context = contextProvider.getContext(); ForwardingStreamObserver<QueryEventsResponse> responseObserver = new ForwardingStreamObserver<>(logger, "queryEvents", responseObserver0); return new StreamObserver<QueryEventsRequest>() { private final AtomicReference<StreamObserver<QueryEventsRequest>> requestObserver = new AtomicReference<>(); @Override public void onNext(QueryEventsRequest request) { if (requestObserver.get() == null) { EventStore eventStore = eventStoreLocator.getEventStore(context, request.getForceReadFromLeader()); if (eventStore == null) { responseObserver.onError(new MessagingPlatformException(ErrorCode.NO_EVENTSTORE, NO_EVENT_STORE_CONFIGURED + context)); return; } requestObserver.set(eventStore.queryEvents(context, responseObserver)); } try { requestObserver.get().onNext(request); } catch (Exception reason) { logger.warn("{}: Error forwarding request to event store: {}", context, reason.getMessage()); StreamObserverUtils.complete(requestObserver.get()); } } @Override public void onError(Throwable reason) { if (!ExceptionUtils.isCancelled(reason)) { logger.warn("Error on connection from client: {}", reason.getMessage()); } cleanup(); } @Override public void onCompleted() { cleanup(); StreamObserverUtils.complete(responseObserver); } private void cleanup() { StreamObserverUtils.complete(requestObserver.get()); } }; } EventDispatcher(EventStoreLocator eventStoreLocator,
                           ContextProvider contextProvider,
                           MeterFactory meterFactory); StreamObserver<InputStream> appendEvent(StreamObserver<Confirmation> responseObserver); StreamObserver<InputStream> appendEvent(String context, StreamObserver<Confirmation> responseObserver); void appendSnapshot(Event event, StreamObserver<Confirmation> confirmationStreamObserver); void appendSnapshot(String context, Event request, StreamObserver<Confirmation> responseObserver); void listAggregateEvents(GetAggregateEventsRequest request,
                                    StreamObserver<SerializedEvent> responseObserver); void listAggregateEvents(String context, GetAggregateEventsRequest request,
                                    StreamObserver<SerializedEvent> responseObserver); StreamObserver<GetEventsRequest> listEvents(StreamObserver<InputStream> responseObserver); StreamObserver<GetEventsRequest> listEvents(String context, StreamObserver<InputStream> responseObserver); @EventListener void on(TopologyEvents.ApplicationDisconnected applicationDisconnected); long getNrOfEvents(String context); Map<String, Iterable<Long>> eventTrackerStatus(String context); @Override final io.grpc.ServerServiceDefinition bindService(); void getFirstToken(GetFirstTokenRequest request, StreamObserver<TrackingToken> responseObserver0); void getLastToken(GetLastTokenRequest request, StreamObserver<TrackingToken> responseObserver0); void getTokenAt(GetTokenAtRequest request, StreamObserver<TrackingToken> responseObserver0); void readHighestSequenceNr(ReadHighestSequenceNrRequest request,
                                      StreamObserver<ReadHighestSequenceNrResponse> responseObserver0); StreamObserver<QueryEventsRequest> queryEvents(StreamObserver<QueryEventsResponse> responseObserver0); void listAggregateSnapshots(String context, GetAggregateSnapshotsRequest request,
                                       StreamObserver<SerializedEvent> responseObserver); MeterFactory.RateMeter eventRate(String context); MeterFactory.RateMeter snapshotRate(String context); }
EventDispatcher implements AxonServerClientService { public StreamObserver<QueryEventsRequest> queryEvents(StreamObserver<QueryEventsResponse> responseObserver0) { String context = contextProvider.getContext(); ForwardingStreamObserver<QueryEventsResponse> responseObserver = new ForwardingStreamObserver<>(logger, "queryEvents", responseObserver0); return new StreamObserver<QueryEventsRequest>() { private final AtomicReference<StreamObserver<QueryEventsRequest>> requestObserver = new AtomicReference<>(); @Override public void onNext(QueryEventsRequest request) { if (requestObserver.get() == null) { EventStore eventStore = eventStoreLocator.getEventStore(context, request.getForceReadFromLeader()); if (eventStore == null) { responseObserver.onError(new MessagingPlatformException(ErrorCode.NO_EVENTSTORE, NO_EVENT_STORE_CONFIGURED + context)); return; } requestObserver.set(eventStore.queryEvents(context, responseObserver)); } try { requestObserver.get().onNext(request); } catch (Exception reason) { logger.warn("{}: Error forwarding request to event store: {}", context, reason.getMessage()); StreamObserverUtils.complete(requestObserver.get()); } } @Override public void onError(Throwable reason) { if (!ExceptionUtils.isCancelled(reason)) { logger.warn("Error on connection from client: {}", reason.getMessage()); } cleanup(); } @Override public void onCompleted() { cleanup(); StreamObserverUtils.complete(responseObserver); } private void cleanup() { StreamObserverUtils.complete(requestObserver.get()); } }; } EventDispatcher(EventStoreLocator eventStoreLocator,
                           ContextProvider contextProvider,
                           MeterFactory meterFactory); StreamObserver<InputStream> appendEvent(StreamObserver<Confirmation> responseObserver); StreamObserver<InputStream> appendEvent(String context, StreamObserver<Confirmation> responseObserver); void appendSnapshot(Event event, StreamObserver<Confirmation> confirmationStreamObserver); void appendSnapshot(String context, Event request, StreamObserver<Confirmation> responseObserver); void listAggregateEvents(GetAggregateEventsRequest request,
                                    StreamObserver<SerializedEvent> responseObserver); void listAggregateEvents(String context, GetAggregateEventsRequest request,
                                    StreamObserver<SerializedEvent> responseObserver); StreamObserver<GetEventsRequest> listEvents(StreamObserver<InputStream> responseObserver); StreamObserver<GetEventsRequest> listEvents(String context, StreamObserver<InputStream> responseObserver); @EventListener void on(TopologyEvents.ApplicationDisconnected applicationDisconnected); long getNrOfEvents(String context); Map<String, Iterable<Long>> eventTrackerStatus(String context); @Override final io.grpc.ServerServiceDefinition bindService(); void getFirstToken(GetFirstTokenRequest request, StreamObserver<TrackingToken> responseObserver0); void getLastToken(GetLastTokenRequest request, StreamObserver<TrackingToken> responseObserver0); void getTokenAt(GetTokenAtRequest request, StreamObserver<TrackingToken> responseObserver0); void readHighestSequenceNr(ReadHighestSequenceNrRequest request,
                                      StreamObserver<ReadHighestSequenceNrResponse> responseObserver0); StreamObserver<QueryEventsRequest> queryEvents(StreamObserver<QueryEventsResponse> responseObserver0); void listAggregateSnapshots(String context, GetAggregateSnapshotsRequest request,
                                       StreamObserver<SerializedEvent> responseObserver); MeterFactory.RateMeter eventRate(String context); MeterFactory.RateMeter snapshotRate(String context); static final MethodDescriptor<GetEventsRequest, InputStream> METHOD_LIST_EVENTS; static final MethodDescriptor<GetAggregateEventsRequest, SerializedEvent> METHOD_LIST_AGGREGATE_EVENTS; static final MethodDescriptor<GetAggregateSnapshotsRequest, SerializedEvent> METHOD_LIST_AGGREGATE_SNAPSHOTS; static final MethodDescriptor<InputStream, Confirmation> METHOD_APPEND_EVENT; }