@Test public void testReadEventsForAggregateWithEventsExceedingOneBatchAndGapsReturnsTheCompleteStream() { int batchSize = 5; testSubject = createEngine(defaultPersistenceExceptionResolver, new EventSchema(), batchSize); DomainEventMessage<String> testEventOne = createEvent(0); DomainEventMessage<String> testEventTwo = createEvent(1); DomainEventMessage<String> testEventFour = createEvent(3); DomainEventMessage<String> testEventFive = createEvent(4); DomainEventMessage<String> testEventSix = createEvent(5); DomainEventMessage<String> testEventSeven = createEvent(6); DomainEventMessage<String> testEventEight = createEvent(7); testSubject.appendEvents( testEventOne, testEventTwo, testEventFour, testEventFive, testEventSix, testEventSeven, testEventEight ); List<? extends DomainEventMessage<?>> result = testSubject.readEvents(AGGREGATE, 0L).asStream() .collect(toList()); assertEquals(7, result.size()); assertEquals(0, result.get(0).getSequenceNumber()); assertEquals(1, result.get(1).getSequenceNumber()); assertEquals(3, result.get(2).getSequenceNumber()); assertEquals(4, result.get(3).getSequenceNumber()); assertEquals(5, result.get(4).getSequenceNumber()); assertEquals(6, result.get(5).getSequenceNumber()); assertEquals(7, result.get(6).getSequenceNumber()); }
protected PreparedStatement appendEvents(Connection connection, List<? extends EventMessage<?>> events, Serializer serializer) throws SQLException { return appendEvents.build(connection, schema, dataType, events, serializer, this::writeTimestamp); }
JdbcEventStorageEngine extends BatchingEventStorageEngine { protected PreparedStatement appendEvents(Connection connection, List<? extends EventMessage<?>> events, Serializer serializer) throws SQLException { return appendEvents.build(connection, schema, dataType, events, serializer, this::writeTimestamp); } }
JdbcEventStorageEngine extends BatchingEventStorageEngine { protected PreparedStatement appendEvents(Connection connection, List<? extends EventMessage<?>> events, Serializer serializer) throws SQLException { return appendEvents.build(connection, schema, dataType, events, serializer, this::writeTimestamp); } protected  JdbcEventStorageEngine(Builder builder); }
JdbcEventStorageEngine extends BatchingEventStorageEngine { protected PreparedStatement appendEvents(Connection connection, List<? extends EventMessage<?>> events, Serializer serializer) throws SQLException { return appendEvents.build(connection, schema, dataType, events, serializer, this::writeTimestamp); } protected  JdbcEventStorageEngine(Builder builder); static Builder builder(); void createSchema(EventTableFactory schemaFactory); @Override Optional<Long> lastSequenceNumberFor(String aggregateIdentifier); @Override TrackingToken createTailToken(); @Override TrackingToken createHeadToken(); @Override TrackingToken createTokenAt(Instant dateTime); @Deprecated void setGapTimeout(int gapTimeout); @Deprecated void setGapCleaningThreshold(int gapCleaningThreshold); }
JdbcEventStorageEngine extends BatchingEventStorageEngine { protected PreparedStatement appendEvents(Connection connection, List<? extends EventMessage<?>> events, Serializer serializer) throws SQLException { return appendEvents.build(connection, schema, dataType, events, serializer, this::writeTimestamp); } protected  JdbcEventStorageEngine(Builder builder); static Builder builder(); void createSchema(EventTableFactory schemaFactory); @Override Optional<Long> lastSequenceNumberFor(String aggregateIdentifier); @Override TrackingToken createTailToken(); @Override TrackingToken createHeadToken(); @Override TrackingToken createTokenAt(Instant dateTime); @Deprecated void setGapTimeout(int gapTimeout); @Deprecated void setGapCleaningThreshold(int gapCleaningThreshold);  }