@SlowTest @Test @DisplayName("be able to store events in parallel") void storeInParallel() { int eventCount = 10_000; CompletableFuture<?>[] futures = Stream.generate(GivenEvent::arbitrary) .limit(eventCount) .map(event -> runAsync(() -> waitAndStore(event), executor)) .toArray(CompletableFuture[]::new); CompletableFuture<Void> allFutures = allOf(futures); allFutures.join(); MemoizingObserver<Event> observer = memoizingObserver(); eventStore.read(EventStreamQuery.getDefaultInstance(), observer); assertThat(observer.isCompleted()) .isTrue(); assertThat(observer.responses()) .hasSize(eventCount); }
@Override public void read(EventStreamQuery request, StreamObserver<Event> responseObserver) { checkNotNull(request); checkNotNull(responseObserver); log.readingStart(request, responseObserver); Iterator<Event> eventRecords = iterator(request); while (eventRecords.hasNext()) { Event event = eventRecords.next(); responseObserver.onNext(event); } responseObserver.onCompleted(); log.readingComplete(responseObserver); }
DefaultEventStore extends DefaultRecordBasedRepository<EventId, EEntity, Event> implements EventStore { @Override public void read(EventStreamQuery request, StreamObserver<Event> responseObserver) { checkNotNull(request); checkNotNull(responseObserver); log.readingStart(request, responseObserver); Iterator<Event> eventRecords = iterator(request); while (eventRecords.hasNext()) { Event event = eventRecords.next(); responseObserver.onNext(event); } responseObserver.onCompleted(); log.readingComplete(responseObserver); } }
DefaultEventStore extends DefaultRecordBasedRepository<EventId, EEntity, Event> implements EventStore { @Override public void read(EventStreamQuery request, StreamObserver<Event> responseObserver) { checkNotNull(request); checkNotNull(responseObserver); log.readingStart(request, responseObserver); Iterator<Event> eventRecords = iterator(request); while (eventRecords.hasNext()) { Event event = eventRecords.next(); responseObserver.onNext(event); } responseObserver.onCompleted(); log.readingComplete(responseObserver); } DefaultEventStore(); }
DefaultEventStore extends DefaultRecordBasedRepository<EventId, EEntity, Event> implements EventStore { @Override public void read(EventStreamQuery request, StreamObserver<Event> responseObserver) { checkNotNull(request); checkNotNull(responseObserver); log.readingStart(request, responseObserver); Iterator<Event> eventRecords = iterator(request); while (eventRecords.hasNext()) { Event event = eventRecords.next(); responseObserver.onNext(event); } responseObserver.onCompleted(); log.readingComplete(responseObserver); } DefaultEventStore(); @Override @SuppressWarnings({"OverridingMethodsMustInvokeSuper", "MissingSuperCall"}) // see impl. note void registerWith(BoundedContext context); @Override void append(Event event); @Override void appendAll(Iterable<Event> events); @Override void read(EventStreamQuery request, StreamObserver<Event> responseObserver); }
DefaultEventStore extends DefaultRecordBasedRepository<EventId, EEntity, Event> implements EventStore { @Override public void read(EventStreamQuery request, StreamObserver<Event> responseObserver) { checkNotNull(request); checkNotNull(responseObserver); log.readingStart(request, responseObserver); Iterator<Event> eventRecords = iterator(request); while (eventRecords.hasNext()) { Event event = eventRecords.next(); responseObserver.onNext(event); } responseObserver.onCompleted(); log.readingComplete(responseObserver); } DefaultEventStore(); @Override @SuppressWarnings({"OverridingMethodsMustInvokeSuper", "MissingSuperCall"}) // see impl. note void registerWith(BoundedContext context); @Override void append(Event event); @Override void appendAll(Iterable<Event> events); @Override void read(EventStreamQuery request, StreamObserver<Event> responseObserver);  }