Skip to content

feat(mongoreactive): add journal event persistence #955

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
bercianor wants to merge 1 commit into
base: master
Choose a base branch
from
Open

feat(mongoreactive): add journal event persistence #955

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 2 additions & 0 deletions .gitignore
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -103,3 +103,5 @@ test_*.db-shm
# Generic SQLite artifacts
*.db-wal
*.db-shm
# Local Pi runtime state
.atl/
105 changes: 80 additions & 25 deletions ...itstore/src/main/java/io/flamingock/store/mongodb/reactive/MongoDBReactiveAuditStore.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -21,22 +21,31 @@
import com.mongodb.reactivestreams.client.ClientSession;
import com.mongodb.reactivestreams.client.MongoDatabase;
import io.flamingock.externalsystem.mongodb.reactive.api.MongoDBReactiveExternalSystem;
import io.flamingock.internal.common.core.audit.AuditPersistenceFactory;
import io.flamingock.internal.common.core.audit.AuditReader;
import io.flamingock.internal.common.core.context.ContextResolver;
import io.flamingock.internal.common.core.error.FlamingockException;
import io.flamingock.internal.common.core.feature.Features;
import io.flamingock.internal.core.configuration.community.CommunityConfigurable;
import io.flamingock.internal.core.external.store.CommunityAuditStore;
import io.flamingock.internal.core.external.store.audit.community.CommunityAuditPersistence;
import io.flamingock.internal.core.external.store.lock.community.CommunityLockService;
import io.flamingock.internal.core.journal.JournalEventSequencer;
import io.flamingock.internal.core.journal.JournalEventSequencerFactory;
import io.flamingock.internal.util.Constants;
import io.flamingock.internal.util.FeatureFlag;
import io.flamingock.internal.util.TimeService;
import io.flamingock.internal.util.id.RunnerId;
import io.flamingock.store.mongodb.reactive.internal.MongoDBReactiveAuditPersistence;
import io.flamingock.store.mongodb.reactive.internal.MongoDBReactiveAuditRepository;
import io.flamingock.store.mongodb.reactive.internal.MongoDBReactiveJournalEventStore;
import io.flamingock.store.mongodb.reactive.internal.MongoDBReactiveLockService;

import java.util.Collections;
import java.util.HashSet;
import java.util.Set;

import static io.flamingock.internal.common.mongodb.journal.JournalEventPersistenceConstants.DEFAULT_JOURNAL_STORE_NAME;
import static io.flamingock.internal.util.constants.CommunityPersistenceConstants.DEFAULT_AUDIT_STORE_NAME;
import static io.flamingock.internal.util.constants.CommunityPersistenceConstants.DEFAULT_LOCK_STORE_NAME;

Expand All @@ -46,15 +55,19 @@ public class MongoDBReactiveAuditStore implements CommunityAuditStore {

protected RunnerId runnerId;
private CommunityConfigurable communityConfiguration;
private MongoDBReactiveAuditPersistence persistence;
private CommunityAuditPersistence persistence;
private MongoDBReactiveLockService lockService;
private MongoDatabase database;
private String auditRepositoryName = DEFAULT_AUDIT_STORE_NAME;
private String lockRepositoryName = DEFAULT_LOCK_STORE_NAME;
private String journalRepositoryName = DEFAULT_JOURNAL_STORE_NAME;
private ReadConcern readConcern = ReadConcern.MAJORITY;
private ReadPreference readPreference = ReadPreference.primary();
private WriteConcern writeConcern = WriteConcern.MAJORITY.withJournal(true);
private boolean autoCreate = true;
private MongoDBReactiveAuditRepository auditRepository;
private MongoDBReactiveJournalEventStore journalEventStore;
private JournalEventSequencerFactory journalEventSequencerFactory;

private MongoDBReactiveAuditStore(MongoDBReactiveExternalSystem mongoDBTargetSystem) {
this.mongoDBTargetSystem = mongoDBTargetSystem;
Expand Down Expand Up @@ -89,6 +102,11 @@ public MongoDBReactiveAuditStore withLockRepositoryName(String lockRepositoryNam
return this;
}

public MongoDBReactiveAuditStore withJournalRepositoryName(String journalRepositoryName) {
this.journalRepositoryName = journalRepositoryName;
return this;
}

public MongoDBReactiveAuditStore withReadConcern(ReadConcern readConcern) {
this.readConcern = readConcern;
return this;
Expand All @@ -114,40 +132,57 @@ public void initialize(ContextResolver baseContext) {
runnerId = baseContext.getRequiredDependencyValue(RunnerId.class);
communityConfiguration = baseContext.getRequiredDependencyValue(CommunityConfigurable.class);
database = mongoDBTargetSystem.getMongoDatabase();
this.validate();
this.validate();

auditRepository = new MongoDBReactiveAuditRepository(
database, auditRepositoryName, readConcern, readPreference, writeConcern);
journalEventStore = new MongoDBReactiveJournalEventStore(
database, journalRepositoryName, readConcern, readPreference, writeConcern);
journalEventSequencerFactory = new JournalEventSequencerFactory(journalEventStore);

lockService = new MongoDBReactiveLockService(
database,
lockRepositoryName,
readConcern,
readPreference,
writeConcern,
TimeService.getDefault()
);
lockService.initialize(autoCreate);
}

@Override
public synchronized CommunityAuditPersistence getPersistence() {
if (persistence == null) {
persistence = new MongoDBReactiveAuditPersistence(
public AuditPersistenceFactory<CommunityAuditPersistence> getPersistenceFactory() {
return stageId -> {
auditRepository.initialize(autoCreate);
if (isJournalEventsEnabled()) {
journalEventStore.initialize(autoCreate);
}
JournalEventSequencer journalEventSequencer = journalEventSequencerFactory.forStream(stageId);
MongoDBReactiveAuditPersistence stagePersistence = new MongoDBReactiveAuditPersistence(
communityConfiguration,
database,
auditRepositoryName,
readConcern,
readPreference,
writeConcern,
auditRepository,
journalEventStore,
journalEventSequencer,
mongoDBTargetSystem.getTxWrapper(),
autoCreate
);
persistence.initialize(runnerId);
}
return persistence;
stagePersistence.initialize(runnerId);
if (persistence == null) {
persistence = stagePersistence;
}
return stagePersistence;
};
}

@Override
public synchronized CommunityLockService getLockService() {
if (lockService == null) {
lockService = new MongoDBReactiveLockService(
database,
lockRepositoryName,
readConcern,
readPreference,
writeConcern,
TimeService.getDefault()
);
lockService.initialize(autoCreate);
public AuditReader getAuditReader() {
auditRepository.initialize(autoCreate);
return () -> auditRepository.getAuditHistory();
}

}
@Override
public synchronized CommunityLockService getLockService() {
return lockService;
}

Expand All @@ -165,10 +200,22 @@ private void validate() {
throw new FlamingockException("The 'lockRepositoryName' property is required.");
}

if (journalRepositoryName == null || journalRepositoryName.trim().isEmpty()) {
throw new FlamingockException("The 'journalRepositoryName' property is required.");
}

if (auditRepositoryName.trim().equalsIgnoreCase(lockRepositoryName.trim())) {
throw new FlamingockException("The 'auditRepositoryName' and 'lockRepositoryName' properties must not be the same.");
}

if (journalRepositoryName.trim().equalsIgnoreCase(auditRepositoryName.trim())) {
throw new FlamingockException("The 'journalRepositoryName' and 'auditRepositoryName' properties must not be the same.");
}

if (journalRepositoryName.trim().equalsIgnoreCase(lockRepositoryName.trim())) {
throw new FlamingockException("The 'journalRepositoryName' and 'lockRepositoryName' properties must not be the same.");
}

if (readConcern == null) {
throw new FlamingockException("The 'readConcern' property is required.");
}
Expand All @@ -182,6 +229,14 @@ private void validate() {
}
}

private static boolean isJournalEventsEnabled() {
try {
return FeatureFlag.isEnabled(Features.JOURNAL_EVENTS, false);
} catch (RuntimeException exception) {
return false;
}
}

@Override
public Set<Class<?>> getNonGuardedTypes() {
return new HashSet<>(Collections.singletonList(ClientSession.class));
Expand Down
91 changes: 69 additions & 22 deletions ...n/java/io/flamingock/store/mongodb/reactive/internal/MongoDBReactiveAuditPersistence.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -21,59 +21,106 @@
import com.mongodb.reactivestreams.client.ClientSession;
import com.mongodb.reactivestreams.client.MongoDatabase;
import io.flamingock.internal.common.core.audit.AuditEntry;
import io.flamingock.internal.common.core.context.RuntimeContext;
import io.flamingock.internal.common.core.feature.Features;
import io.flamingock.internal.common.core.journal.JournalEvent;
import io.flamingock.internal.common.core.transaction.TransactionWrapper;
import io.flamingock.internal.core.context.BasicRuntimeContext;
import io.flamingock.internal.core.configuration.community.CommunityConfigurable;
import io.flamingock.internal.core.external.store.audit.community.AbstractCommunityAuditPersistence;
import io.flamingock.internal.core.journal.JournalEventSequencer;
import io.flamingock.internal.util.FeatureFlag;
import io.flamingock.internal.util.Result;
import io.flamingock.internal.util.id.RunnerId;

import java.util.Collections;
import java.util.HashSet;
import java.util.List;
import java.util.Set;

import static io.flamingock.internal.common.mongodb.journal.JournalEventPersistenceConstants.DEFAULT_JOURNAL_STORE_NAME;

public class MongoDBReactiveAuditPersistence extends AbstractCommunityAuditPersistence {

private MongoDBReactiveAuditor auditor;
private final MongoDatabase database;
private final String auditCollectionName;
private final ReadConcern readConcern;
private final ReadPreference readPreference;
private final WriteConcern writeConcern;
private final MongoDBReactiveAuditRepository auditRepository;
private final MongoDBReactiveJournalEventStore journalEventStore;
private final JournalEventSequencer journalEventSequencer;
private final TransactionWrapper txWrapper;
private final boolean autoCreate;

public MongoDBReactiveAuditPersistence(CommunityConfigurable localConfiguration,
MongoDBReactiveAuditRepository auditRepository,
MongoDBReactiveJournalEventStore journalEventStore,
JournalEventSequencer journalEventSequencer,
TransactionWrapper txWrapper,
boolean autoCreate) {
super(localConfiguration);
this.auditRepository = auditRepository;
this.journalEventStore = journalEventStore;
this.journalEventSequencer = journalEventSequencer;
this.txWrapper = txWrapper;
this.autoCreate = autoCreate;
}

/**
* Backward-compatible constructor for callers that only need the historical audit path.
*/
public MongoDBReactiveAuditPersistence(CommunityConfigurable localConfiguration,
MongoDatabase database,
String auditCollectionName,
ReadConcern readConcern,
ReadPreference readPreference,
WriteConcern writeConcern,
boolean autoCreate) {
super(localConfiguration);
this.database = database;
this.auditCollectionName = auditCollectionName;
this.readConcern = readConcern;
this.readPreference = readPreference;
this.writeConcern = writeConcern;
this.autoCreate = autoCreate;
this(
localConfiguration,
new MongoDBReactiveAuditRepository(database, auditCollectionName, readConcern, readPreference, writeConcern),
new MongoDBReactiveJournalEventStore(database, DEFAULT_JOURNAL_STORE_NAME,
readConcern, readPreference, writeConcern),
null,
null,
autoCreate);
}

@Override
protected void doInitialize(RunnerId runnerId) {
//Auditor
auditor = new MongoDBReactiveAuditor(database, auditCollectionName, readConcern, readPreference, writeConcern);
auditor.initialize(autoCreate);
auditRepository.initialize(autoCreate);
if (isJournalEventsEnabled()) {
journalEventStore.initialize(autoCreate);
}
}


@Override
public List<AuditEntry> getAuditHistory() {
return auditor.getAuditHistory();
return auditRepository.getAuditHistory();
}

@Override
public Result writeEntry(AuditEntry auditEntry) {
return auditor.writeEntry(auditEntry);
if (!isJournalEventsEnabled()) {
return auditRepository.append(auditEntry);
}

if (journalEventStore == null || journalEventSequencer == null || txWrapper == null) {
throw new IllegalStateException("MongoDB reactive journal writes require a transaction wrapper and sequencer");
}

RuntimeContext baseContext = new BasicRuntimeContext("write-changeState-" + auditEntry.getChangeId());
Result result = txWrapper.wrapInTransaction(baseContext, runtimeContext -> {
ClientSession clientSession = runtimeContext.getContext().getRequiredDependencyValue(ClientSession.class);
JournalEvent<AuditEntry> journalEvent = journalEventSequencer.newEvent(auditEntry);
journalEventStore.append(clientSession, journalEvent);
return auditRepository.save(clientSession, auditEntry);
});

// Result cannot represent FailedStep. The transaction wrapper has therefore committed successfully
// whenever control reaches this line; only then is the in-memory stream position spent.
journalEventSequencer.confirm();
return result;
}

private static boolean isJournalEventsEnabled() {
try {
return FeatureFlag.isEnabled(Features.JOURNAL_EVENTS, false);
} catch (RuntimeException exception) {
return false;
}
}
}
Loading
Loading