### Question:
EventDispatcher implements AxonServerClientService { public StreamObserver<InputStream> appendEvent(StreamObserver<Confirmation> responseObserver) { return appendEvent(contextProvider.getContext(), new ForwardingStreamObserver<>(logger, "appendEvent", 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 appendEvent() { FakeStreamObserver<Confirmation> responseObserver = new FakeStreamObserver<>(); StreamObserver<InputStream> inputStream = testSubject.appendEvent(responseObserver); inputStream.onNext(dummyEvent()); assertTrue(responseObserver.values().isEmpty()); inputStream.onCompleted(); verify( appendEventConnection).onCompleted(); }
@Test public void appendEventRollback() { FakeStreamObserver<Confirmation> responseObserver = new FakeStreamObserver<>(); StreamObserver<InputStream> inputStream = testSubject.appendEvent(responseObserver); inputStream.onNext(dummyEvent()); assertTrue(responseObserver.values().isEmpty()); inputStream.onError(new Throwable()); assertTrue(responseObserver.errors().isEmpty()); verify(appendEventConnection).onError(any()); }