From d4f14486f8a5219233303f05afd98e976f112348 Mon Sep 17 00:00:00 2001 From: Steffen Schaefer Date: Tue, 13 Aug 2019 08:58:30 +0200 Subject: [PATCH] Bug 5085: Initial setup of replication --- .../impl/ReplicatingSharedSailingData.java | 5 +- .../server/impl/SharedSailingDataImpl.java | 136 ++++++++++++++++++ 2 files changed, 140 insertions(+), 1 deletion(-) diff --git a/java/com.sap.sailing.server/src/com/sap/sailing/server/impl/ReplicatingSharedSailingData.java b/java/com.sap.sailing.server/src/com/sap/sailing/server/impl/ReplicatingSharedSailingData.java index 88c65deba67..af22b74e4b1 100644 --- a/java/com.sap.sailing.server/src/com/sap/sailing/server/impl/ReplicatingSharedSailingData.java +++ b/java/com.sap.sailing.server/src/com/sap/sailing/server/impl/ReplicatingSharedSailingData.java @@ -1,7 +1,10 @@ package com.sap.sailing.server.impl; import com.sap.sailing.domain.sharedsailingdata.SharedSailingData; +import com.sap.sse.replication.OperationWithResult; +import com.sap.sse.replication.ReplicableWithObjectInputStream; -public interface ReplicatingSharedSailingData extends SharedSailingData { +public interface ReplicatingSharedSailingData extends SharedSailingData, + ReplicableWithObjectInputStream> { } diff --git a/java/com.sap.sailing.server/src/com/sap/sailing/server/impl/SharedSailingDataImpl.java b/java/com.sap.sailing.server/src/com/sap/sailing/server/impl/SharedSailingDataImpl.java index 7ec622af545..38349180ced 100644 --- a/java/com.sap.sailing.server/src/com/sap/sailing/server/impl/SharedSailingDataImpl.java +++ b/java/com.sap.sailing.server/src/com/sap/sailing/server/impl/SharedSailingDataImpl.java @@ -1,6 +1,14 @@ package com.sap.sailing.server.impl; +import java.io.IOException; +import java.io.InputStream; +import java.io.ObjectInputStream; +import java.io.ObjectOutputStream; +import java.io.Serializable; +import java.net.MalformedURLException; +import java.util.HashSet; import java.util.Map; +import java.util.Set; import java.util.UUID; import java.util.concurrent.ConcurrentHashMap; import java.util.logging.Logger; @@ -20,6 +28,13 @@ import com.sap.sailing.domain.persistence.racelog.tracking.DeviceIdentifierMongo import com.sap.sse.common.TimePoint; import com.sap.sse.common.TypeBasedServiceFinder; import com.sap.sse.common.Util; +import com.sap.sse.replication.OperationExecutionListener; +import com.sap.sse.replication.OperationWithResult; +import com.sap.sse.replication.OperationWithResultWithIdWrapper; +import com.sap.sse.replication.OperationsToMasterSender; +import com.sap.sse.replication.OperationsToMasterSendingQueue; +import com.sap.sse.replication.ReplicationMasterDescriptor; +import com.sap.sse.replication.ReplicationService; import com.sap.sse.util.ClearStateTestSupport; public class SharedSailingDataImpl implements ReplicatingSharedSailingData, ClearStateTestSupport { @@ -42,6 +57,10 @@ public class SharedSailingDataImpl implements ReplicatingSharedSailingData, Clea @Override public void clearState() throws Exception { + removeAll(); + } + + private void removeAll() { markPropertiesById.clear(); } @@ -172,5 +191,122 @@ public class SharedSailingDataImpl implements ReplicatingSharedSailingData, Clea // TODO Auto-generated method stub } + + // Replication related methods and fields + private final ConcurrentHashMap, OperationExecutionListener> operationExecutionListeners = new ConcurrentHashMap<>(); + private ThreadLocal currentlyFillingFromInitialLoad = ThreadLocal.withInitial(() -> false); + private ThreadLocal currentlyApplyingOperationReceivedFromMaster = ThreadLocal.withInitial(() -> false); + private final Set> operationsSentToMasterForReplication = new HashSet<>(); + private ReplicationMasterDescriptor master; + + /** + * This field is expected to be set by the {@link ReplicationService} once it has "adopted" this replicable. + * The {@link ReplicationService} "injects" this service so it can be used here as a delegate for the + * {@link OperationsToMasterSendingQueue#scheduleForSending(OperationWithResult, OperationsToMasterSender)} + * method. + */ + private OperationsToMasterSendingQueue unsentOperationsToMasterSender; + + + @Override + public Serializable getId() { + return getClass().getName(); + } + + @Override + public ReplicationMasterDescriptor getMasterDescriptor() { + return master; + } + + @Override + public void startedReplicatingFrom(ReplicationMasterDescriptor master) { + this.master = master; + } + + @Override + public void stoppedReplicatingFrom(ReplicationMasterDescriptor master) { + this.master = null; + } + + @Override + public Iterable> getOperationExecutionListeners() { + return operationExecutionListeners.keySet(); + } + + @Override + public void addOperationExecutionListener( + OperationExecutionListener listener) { + this.operationExecutionListeners.put(listener, listener); + } + + @Override + public void removeOperationExecutionListener( + OperationExecutionListener listener) { + this.operationExecutionListeners.remove(listener); + } + + @Override + public boolean isCurrentlyFillingFromInitialLoad() { + return currentlyFillingFromInitialLoad.get(); + } + + @Override + public void setCurrentlyFillingFromInitialLoad(boolean currentlyFillingFromInitialLoad) { + this.currentlyFillingFromInitialLoad.set(currentlyFillingFromInitialLoad); + } + + @Override + public boolean isCurrentlyApplyingOperationReceivedFromMaster() { + return currentlyApplyingOperationReceivedFromMaster.get(); + } + + @Override + public void setCurrentlyApplyingOperationReceivedFromMaster(boolean currentlyApplyingOperationReceivedFromMaster) { + this.currentlyApplyingOperationReceivedFromMaster.set(currentlyApplyingOperationReceivedFromMaster); + } + + @Override + public void addOperationSentToMasterForReplication( + OperationWithResultWithIdWrapper operationWithResultWithIdWrapper) { + this.operationsSentToMasterForReplication.add(operationWithResultWithIdWrapper); + } + + @Override + public boolean hasSentOperationToMaster( + OperationWithResult operation) { + return operationsSentToMasterForReplication.remove(operation); + } + + @Override + public ObjectInputStream createObjectInputStreamResolvingAgainstCache(InputStream is) throws IOException { + return new ObjectInputStream(is); + } + + @Override + public synchronized void initiallyFillFromInternal(ObjectInputStream is) + throws IOException, ClassNotFoundException, InterruptedException { + } + + @Override + public void serializeForInitialReplicationInternal(ObjectOutputStream objectOutputStream) throws IOException { + } + + @Override + public synchronized void clearReplicaState() throws MalformedURLException, IOException, InterruptedException { + removeAll(); + } + + @Override + public void setUnsentOperationToMasterSender(OperationsToMasterSendingQueue service) { + this.unsentOperationsToMasterSender = service; + } + + @Override + public , T> void scheduleForSending( + OperationWithResult operationWithResult, OperationsToMasterSender sender) { + if (unsentOperationsToMasterSender != null) { + unsentOperationsToMasterSender.scheduleForSending(operationWithResult, sender); + } + } }