refactor: journal event foundation (#942)

Author: dieppa

cloud/flamingock-cloud/src/main/java/io/flamingock/cloud/CloudAuditPersistenceImpl.java

@@ -16,6 +16,7 @@
 package io.flamingock.cloud;
 
 import io.flamingock.internal.common.core.audit.AuditEntry;
+import io.flamingock.internal.common.core.audit.AuditWriter;
 import io.flamingock.internal.common.core.audit.issue.AuditEntryIssue;
 import io.flamingock.internal.common.core.context.ContextContributor;
 import io.flamingock.internal.util.Result;
@@ -24,7 +25,6 @@
 import io.flamingock.internal.util.id.ServiceId;
 import io.flamingock.internal.core.external.store.audit.cloud.CloudAuditPersistence;
 import io.flamingock.internal.common.core.context.ContextInjectable;
-import io.flamingock.internal.core.external.store.audit.LifecycleAuditWriter;
 import io.flamingock.internal.core.plan.ExecutionPlanner;
 
 import java.util.List;
@@ -36,15 +36,15 @@ public final class CloudAuditPersistenceImpl implements CloudAuditPersistence, C
 
     private final ServiceId serviceId;
 
-    private final LifecycleAuditWriter auditWriter;
+    private final AuditWriter auditWriter;
 
     private final ExecutionPlanner executionPlanner;
     private final String jwt;
 
     CloudAuditPersistenceImpl(EnvironmentId environmentId,
                               ServiceId serviceId,
                               String jwt,
-                              LifecycleAuditWriter auditWriter,
+                              AuditWriter auditWriter,
                               ExecutionPlanner executionPlanner,
                               Runnable closer) {
         this.environmentId =environmentId;

cloud/flamingock-cloud/src/main/java/io/flamingock/cloud/CloudAuditStoreImpl.java

@@ -15,6 +15,7 @@
  */
 package io.flamingock.cloud;
 
+import io.flamingock.internal.common.core.audit.AuditWriter;
 import io.flamingock.internal.util.Constants;
 import io.flamingock.internal.util.JsonObjectMapper;
 import io.flamingock.internal.util.id.RunnerId;
@@ -35,7 +36,6 @@
 import io.flamingock.cloud.planner.CloudExecutionPlanner;
 import io.flamingock.cloud.planner.client.ExecutionPlannerClient;
 import io.flamingock.cloud.planner.client.HttpExecutionPlannerClient;
-import io.flamingock.internal.core.external.store.audit.LifecycleAuditWriter;
 import io.flamingock.internal.core.external.targets.TransactionalTargetSystem;
 import io.flamingock.internal.core.external.targets.TargetSystemManager;
 import io.flamingock.internal.core.external.targets.mark.TargetSystemAuditMarker;
@@ -107,7 +107,7 @@ private CloudAuditPersistenceImpl buildPersistence(RunnerId runnerId,
         EnvironmentId environmentId = EnvironmentId.fromLong(authResponse.getEnvironmentId());
         ServiceId serviceId = ServiceId.fromLong(authResponse.getServiceId());
 
-        LifecycleAuditWriter auditWriter = new HtttpAuditWriter(
+        AuditWriter auditWriter = new HtttpAuditWriter(
                 cloudConfiguration.getHost(),
                 environmentId,
                 serviceId,

cloud/flamingock-cloud/src/main/java/io/flamingock/cloud/audit/CloudAuditWriter.java

@@ -15,25 +15,12 @@
  */
 package io.flamingock.cloud.audit;
 
+import io.flamingock.internal.common.core.audit.AuditWriter;
 import io.flamingock.internal.util.Result;
-import io.flamingock.internal.core.external.store.audit.LifecycleAuditWriter;
 import io.flamingock.internal.core.external.store.audit.domain.ExecutionAuditContextBundle;
 import io.flamingock.internal.core.external.store.audit.domain.RollbackAuditContextBundle;
 import io.flamingock.internal.core.external.store.audit.domain.StartExecutionAuditContextBundle;
 
-public interface CloudAuditWriter extends LifecycleAuditWriter {
+public interface CloudAuditWriter extends AuditWriter {
 
-    default Result writeStartExecution(StartExecutionAuditContextBundle auditContextBundle) {
-        return Result.OK();//TODO remove this
-//        return writeEntry(AuditEntryMapper.map(auditContextBundle));
-    }
-
-
-    default Result writeExecution(ExecutionAuditContextBundle auditContextBundle) {
-        return writeEntry(auditContextBundle.toAuditEntry());
-    }
-
-    default Result writeRollback(RollbackAuditContextBundle auditContextBundle) {
-        return writeEntry(auditContextBundle.toAuditEntry());
-    }
 }

community/flamingock-couchbase-auditstore/src/main/java/io/flamingock/store/couchbase/internal/CouchbaseAuditor.java

@@ -25,10 +25,10 @@
 import com.couchbase.client.java.kv.UpsertOptions;
 import io.flamingock.internal.common.core.audit.AuditEntry;
 import io.flamingock.internal.common.core.audit.AuditReader;
+import io.flamingock.internal.common.core.audit.AuditWriter;
 import io.flamingock.internal.common.couchbase.CouchbaseAuditMapper;
 import io.flamingock.internal.common.couchbase.CouchbaseCollectionHelper;
 import io.flamingock.internal.common.couchbase.CouchbaseCollectionInitializator;
-import io.flamingock.internal.core.external.store.audit.LifecycleAuditWriter;
 import io.flamingock.internal.util.Result;
 import io.flamingock.internal.util.log.FlamingockLoggerFactory;
 import org.slf4j.Logger;
@@ -37,7 +37,7 @@
 import java.util.stream.Collectors;
 
 
-public class CouchbaseAuditor implements LifecycleAuditWriter, AuditReader {
+public class CouchbaseAuditor implements AuditWriter, AuditReader {
 
     private static final Logger logger = FlamingockLoggerFactory.getLogger("CouchbaseAuditor");
 

community/flamingock-dynamodb-auditstore/src/main/java/io/flamingock/store/dynamodb/internal/DynamoDBAuditor.java

@@ -15,7 +15,7 @@
  */
 package io.flamingock.store.dynamodb.internal;
 
-import io.flamingock.internal.core.external.store.audit.LifecycleAuditWriter;
+import io.flamingock.internal.common.core.audit.AuditWriter;
 import io.flamingock.internal.core.external.store.audit.community.CommunityAuditReader;
 import io.flamingock.internal.util.dynamodb.entities.AuditEntryEntity;
 import io.flamingock.internal.common.core.audit.AuditEntry;
@@ -36,7 +36,7 @@
 
 import static java.util.Collections.emptyList;
 
-public class DynamoDBAuditor implements LifecycleAuditWriter, CommunityAuditReader {
+public class DynamoDBAuditor implements AuditWriter, CommunityAuditReader {
 
     private static final Logger logger = FlamingockLoggerFactory.getLogger("DynamoAuditor");
 

community/flamingock-mongodb-reactive-auditstore/src/main/java/io/flamingock/store/mongodb/reactive/internal/MongoDBReactiveAuditor.java

@@ -25,11 +25,11 @@
 import com.mongodb.reactivestreams.client.MongoDatabase;
 import io.flamingock.internal.common.core.audit.AuditEntry;
 import io.flamingock.internal.common.core.audit.AuditReader;
+import io.flamingock.internal.common.core.audit.AuditWriter;
 import io.flamingock.internal.common.mongodb.CollectionInitializator;
 import io.flamingock.internal.common.mongodb.MongoDBAuditMapper;
 import io.flamingock.internal.common.mongodb.MongoDBReactiveCollectionHelper;
 import io.flamingock.internal.common.mongodb.MongoDBDocumentHelper;
-import io.flamingock.internal.core.external.store.audit.LifecycleAuditWriter;
 import io.flamingock.internal.util.Result;
 import io.flamingock.internal.util.log.FlamingockLoggerFactory;
 import io.flamingock.reactive.util.PublisherSync;
@@ -44,7 +44,7 @@
 import static io.flamingock.internal.util.constants.AuditEntryFieldConstants.KEY_EXECUTION_ID;
 import static io.flamingock.internal.util.constants.AuditEntryFieldConstants.KEY_STATE;
 
-public class MongoDBReactiveAuditor implements LifecycleAuditWriter, AuditReader {
+public class MongoDBReactiveAuditor implements AuditWriter, AuditReader {
 
     private static final Logger logger = FlamingockLoggerFactory.getLogger("MongoDBReactiveAuditor");
 

community/flamingock-mongodb-sync-auditstore/src/main/java/io/flamingock/store/mongodb/sync/MongoDBSyncAuditStore.java

@@ -32,6 +32,7 @@
 import io.flamingock.store.mongodb.sync.internal.MongoDBSyncLockService;
 import io.flamingock.externalsystem.mongodb.api.MongoDBExternalSystem;
 
+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;
 
@@ -46,6 +47,7 @@ public class MongoDBSyncAuditStore implements CommunityAuditStore {
     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);
@@ -85,6 +87,11 @@ public MongoDBSyncAuditStore withLockRepositoryName(String lockRepositoryName) {
         return this;
     }
 
+    public MongoDBSyncAuditStore withJournalRepositoryName(String journalRepositoryName) {
+        this.journalRepositoryName = journalRepositoryName;
+        return this;
+    }
+
     public MongoDBSyncAuditStore withReadConcern(ReadConcern readConcern) {
         this.readConcern = readConcern;
         return this;
@@ -120,6 +127,7 @@ public synchronized CommunityAuditPersistence getPersistence() {
                     communityConfiguration,
                     database,
                     auditRepositoryName,
+                    journalRepositoryName,
                     readConcern,
                     readPreference,
                     writeConcern,
@@ -161,10 +169,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.");
         }

community/flamingock-mongodb-sync-auditstore/src/main/java/io/flamingock/store/mongodb/sync/internal/MongoDBSyncAuditPersistence.java

@@ -34,8 +34,10 @@
 public class MongoDBSyncAuditPersistence extends AbstractCommunityAuditPersistence {
 
     private MongoDBSyncAuditor auditor;
+    private MongoDBSyncJournalEventStore journalEventStore;
     private final MongoDatabase database;
     private final String auditCollectionName;
+    private final String journalCollectionName;
     private final ReadConcern readConcern;
     private final ReadPreference readPreference;
     private final WriteConcern writeConcern;
@@ -45,13 +47,15 @@ public class MongoDBSyncAuditPersistence extends AbstractCommunityAuditPersisten
     public MongoDBSyncAuditPersistence(CommunityConfigurable localConfiguration,
                                      MongoDatabase database,
                                      String auditCollectionName,
+                                     String journalCollectionName,
                                      ReadConcern readConcern,
                                      ReadPreference readPreference,
                                      WriteConcern writeConcern,
                                      boolean autoCreate) {
         super(localConfiguration);
         this.database = database;
         this.auditCollectionName = auditCollectionName;
+        this.journalCollectionName = journalCollectionName;
         this.readConcern = readConcern;
         this.readPreference = readPreference;
         this.writeConcern = writeConcern;
@@ -63,6 +67,9 @@ protected void doInitialize(RunnerId runnerId) {
         //Auditor
         auditor = new MongoDBSyncAuditor(database, auditCollectionName, readConcern, readPreference, writeConcern);
         auditor.initialize(autoCreate);
+        //Journal
+        journalEventStore = new MongoDBSyncJournalEventStore(database, journalCollectionName, readConcern, readPreference, writeConcern);
+        journalEventStore.initialize(autoCreate);
     }
 
     @Deprecated

community/flamingock-mongodb-sync-auditstore/src/main/java/io/flamingock/store/mongodb/sync/internal/MongoDBSyncAuditor.java

@@ -25,11 +25,11 @@
 import com.mongodb.client.result.UpdateResult;
 import io.flamingock.internal.common.core.audit.AuditEntry;
 import io.flamingock.internal.common.core.audit.AuditReader;
+import io.flamingock.internal.common.core.audit.AuditWriter;
 import io.flamingock.internal.common.mongodb.CollectionInitializator;
 import io.flamingock.internal.common.mongodb.MongoDBAuditMapper;
 import io.flamingock.internal.common.mongodb.MongoDBSyncCollectionHelper;
 import io.flamingock.internal.common.mongodb.MongoDBDocumentHelper;
-import io.flamingock.internal.core.external.store.audit.LifecycleAuditWriter;
 import io.flamingock.internal.util.Result;
 import io.flamingock.internal.util.log.FlamingockLoggerFactory;
 import org.bson.Document;
@@ -44,7 +44,7 @@
 import static io.flamingock.internal.util.constants.AuditEntryFieldConstants.KEY_EXECUTION_ID;
 import static io.flamingock.internal.util.constants.AuditEntryFieldConstants.KEY_STATE;
 
-public class MongoDBSyncAuditor implements LifecycleAuditWriter, AuditReader {
+public class MongoDBSyncAuditor implements AuditWriter, AuditReader {
 
     private static final Logger logger = FlamingockLoggerFactory.getLogger("MongoDBSyncAuditor");