### Question:
EventDispatcher implements AxonServerClientService { public StreamObserver<GetEventsRequest> listEvents(StreamObserver<InputStream> responseObserver) { return listEvents(contextProvider.getContext(), responseObserver); } 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; }### Answer:
@Test public void listEvents() { FakeStreamObserver<InputStream> responseObserver = new FakeStreamObserver<>(); AtomicReference<StreamObserver<InputStream>> eventStoreOutputStreamRef = new AtomicReference<>(); StreamObserver<GetEventsRequest> eventStoreResponseStream = new StreamObserver<GetEventsRequest>() { @Override public void onNext(GetEventsRequest o) { StreamObserver<InputStream> responseStream = eventStoreOutputStreamRef.get(); responseStream.onNext(new ByteArrayInputStream(EventWithToken.newBuilder().build().toByteArray())); responseStream.onCompleted(); } @Override public void onError(Throwable throwable) { } @Override public void onCompleted() { } }; when(eventStoreClient.listEvents(any(), any(StreamObserver.class))).then(a -> { eventStoreOutputStreamRef.set((StreamObserver<InputStream>) a.getArguments()[1]); return eventStoreResponseStream; }); StreamObserver<GetEventsRequest> inputStream = testSubject.listEvents(responseObserver); inputStream.onNext(GetEventsRequest.newBuilder() .setClientId("sampleClient") .build()); verify(eventStoreLocator).getEventStore(Topology.DEFAULT_CONTEXT, false); assertEquals(1, responseObserver.values().size()); responseObserver = new FakeStreamObserver<>(); inputStream = testSubject.listEvents(responseObserver); inputStream.onNext(GetEventsRequest.newBuilder() .setForceReadFromLeader(true) .setClientId("sampleClient") .build()); verify(eventStoreLocator).getEventStore(Topology.DEFAULT_CONTEXT, true); assertEquals(1, responseObserver.completedCount()); testSubject.on(new TopologyEvents.ApplicationDisconnected(Topology.DEFAULT_CONTEXT, "myComponent", "sampleClient")); assertTrue(testSubject.eventTrackerStatus(Topology.DEFAULT_CONTEXT).isEmpty()); }