another interims commit; Peer and Transformer can work with Operation; OperationalTransformer uses OperationWithTransformationSupport

java/com.sap.sailing.server.replication.test/src/com/sap/sailing/server/replication/test/AbstractServerReplicationTest.java

@@ -35,6 +35,7 @@ import com.sap.sailing.domain.persistence.media.MediaDBFactory;
 import com.sap.sailing.domain.racelog.tracking.EmptyGPSFixStore;
 import com.sap.sailing.domain.tracking.impl.EmptyWindStore;
 import com.sap.sailing.server.RacingEventService;
+import com.sap.sailing.server.RacingEventServiceOperation;
 import com.sap.sailing.server.impl.RacingEventServiceImpl;
 import com.sap.sailing.server.replication.ReplicationMasterDescriptor;
 import com.sap.sailing.server.replication.ReplicationService;
@@ -55,7 +56,7 @@ public abstract class AbstractServerReplicationTest {
     protected RacingEventServiceImpl master;
     protected ReplicationServiceTestImpl replicaReplicator;
     private ReplicaDescriptor replicaDescriptor;
-    private ReplicationServiceImpl masterReplicator;
+    private ReplicationServiceImpl<RacingEventService, RacingEventServiceOperation<?>> masterReplicator;
     private ReplicationMasterDescriptor  masterDescriptor;
     
     @Rule public Timeout AbstractTracTracLiveTestTimeout = new Timeout(5 * 60 * 1000); // timeout after 5 minutes
@@ -115,7 +116,7 @@ public abstract class AbstractServerReplicationTest {
                     new DomainFactoryImpl()), mongoObjectFactory, MediaDBFactory.INSTANCE.getMediaDB(mongoDBService), EmptyWindStore.INSTANCE, EmptyGPSFixStore.INSTANCE);
         }
         ReplicationInstancesManager rim = new ReplicationInstancesManager();
-        masterReplicator = new ReplicationServiceImpl(exchangeName, exchangeHost, rim, this.master);
+        masterReplicator = new ReplicationServiceImpl<RacingEventService, RacingEventServiceOperation<?>>(exchangeName, exchangeHost, rim, this.master);
         replicaDescriptor = new ReplicaDescriptor(InetAddress.getLocalHost(), serverUuid, "");
         masterReplicator.registerReplica(replicaDescriptor);
         // connect to exchange host and local server running as master
@@ -166,7 +167,7 @@ public abstract class AbstractServerReplicationTest {
         replicaReplicator.stopToReplicateFromMaster();
     }
     
-    static class ReplicationServiceTestImpl extends ReplicationServiceImpl {
+    static class ReplicationServiceTestImpl extends ReplicationServiceImpl<RacingEventService, RacingEventServiceOperation<?>> {
         protected static final int INITIAL_LOAD_PACKAGE_SIZE = 1024*1024;
         private final DomainFactory resolveAgainst;
         private final RacingEventService master;
@@ -258,12 +259,13 @@ public abstract class AbstractServerReplicationTest {
 //            replicator.setSuspended(false); // resume after initial load
 //        }
 
-        protected Replicator startToReplicateFromButDontYetFetchInitialLoad(ReplicationMasterDescriptor master, boolean startReplicatorSuspended)
+        protected Replicator<RacingEventService, RacingEventServiceOperation<?>> startToReplicateFromButDontYetFetchInitialLoad(ReplicationMasterDescriptor master, boolean startReplicatorSuspended)
                 throws IOException {
             masterReplicationService.registerReplica(replicaDescriptor);
             registerReplicaUuidForMaster(replicaDescriptor.getUuid().toString(), master);
             QueueingConsumer consumer = master.getConsumer();
-            final Replicator replicator = new Replicator(master, this, startReplicatorSuspended, consumer);
+            final Replicator<RacingEventService, RacingEventServiceOperation<?>> replicator = new Replicator<RacingEventService, RacingEventServiceOperation<?>>(
+                    master, this, startReplicatorSuspended, consumer);
             new Thread(replicator).start();
             return replicator;
         }

java/com.sap.sailing.server.replication.test/src/com/sap/sailing/server/replication/test/InitialLoadReplicationObjectIdentityTest.java

@@ -42,6 +42,8 @@ import com.sap.sailing.domain.persistence.media.MediaDBFactory;
 import com.sap.sailing.domain.racelog.tracking.EmptyGPSFixStore;
 import com.sap.sailing.domain.test.TrackBasedTest;
 import com.sap.sailing.domain.tracking.impl.EmptyWindStore;
+import com.sap.sailing.server.RacingEventService;
+import com.sap.sailing.server.RacingEventServiceOperation;
 import com.sap.sailing.server.impl.RacingEventServiceImpl;
 import com.sap.sailing.server.operationaltransformation.AddDefaultRegatta;
 import com.sap.sailing.server.operationaltransformation.AddRaceDefinition;
@@ -140,7 +142,7 @@ public class InitialLoadReplicationObjectIdentityTest extends AbstractServerRepl
         /* fire up replication */
         performReplicationSetup();
         ReplicationMasterDescriptor the_master = replicationDescriptorPair.getB(); /* master descriptor */
-        Replicator replicator = replicationDescriptorPair.getA().startToReplicateFromButDontYetFetchInitialLoad(the_master, /* startReplicatorSuspended */ true);
+        Replicator<RacingEventService, RacingEventServiceOperation<?>> replicator = replicationDescriptorPair.getA().startToReplicateFromButDontYetFetchInitialLoad(the_master, /* startReplicatorSuspended */ true);
         replicationDescriptorPair.getA().initialLoad();
         replicator.setSuspended(false);
         synchronized (replicator) {

java/com.sap.sailing.server.replication.test/src/com/sap/sailing/server/replication/test/PrematureOperationReceiptTest.java

@@ -10,13 +10,15 @@ import org.junit.Test;
 
 import com.sap.sailing.domain.leaderboard.Leaderboard;
 import com.sap.sailing.domain.leaderboard.impl.LowPoint;
+import com.sap.sailing.server.RacingEventService;
+import com.sap.sailing.server.RacingEventServiceOperation;
 import com.sap.sailing.server.operationaltransformation.CreateFlexibleLeaderboard;
 import com.sap.sailing.server.replication.ReplicationMasterDescriptor;
 import com.sap.sailing.server.replication.impl.Replicator;
 import com.sap.sse.common.Util;
 
 public class PrematureOperationReceiptTest extends AbstractServerReplicationTest {
-    private Replicator replicator;
+    private Replicator<RacingEventService, RacingEventServiceOperation<?>> replicator;
     private ReplicationServiceTestImpl replicationService;
 
     /**

java/com.sap.sailing.server.replication.test/src/com/sap/sailing/server/replication/test/TrackRaceReplicationTest.java

@@ -32,6 +32,7 @@ import com.sap.sailing.domain.tracking.DynamicTrackedRace;
 import com.sap.sailing.domain.tracking.RaceHandle;
 import com.sap.sailing.domain.tracking.RaceTrackingConnectivityParameters;
 import com.sap.sailing.domain.tracking.TrackedRace;
+import com.sap.sailing.server.RacingEventService;
 import com.sap.sailing.server.operationaltransformation.AddColumnToLeaderboard;
 import com.sap.sailing.server.operationaltransformation.ConnectTrackedRaceToLeaderboardColumn;
 import com.sap.sailing.server.operationaltransformation.CreateFlexibleLeaderboard;
@@ -40,7 +41,6 @@ import com.sap.sailing.server.replication.OperationExecutionListener;
 import com.sap.sse.common.Util;
 import com.sap.sse.common.impl.MillisecondsTimePoint;
 import com.sap.sse.operationaltransformation.Operation;
-import com.sap.sse.operationaltransformation.OperationWithTransformationSupport;
 
 public class TrackRaceReplicationTest extends AbstractServerReplicationTest {
     private TrackedRace masterTrackedRace;
@@ -71,9 +71,9 @@ public class TrackRaceReplicationTest extends AbstractServerReplicationTest {
         MillisecondsTimePoint startOfTracking = new MillisecondsTimePoint(cal.getTimeInMillis());
         cal.set(2011, 05, 23, 15, 14, 31);
         MillisecondsTimePoint endOfTracking = new MillisecondsTimePoint(cal.getTimeInMillis());
-        master.addOperationExecutionListener(new OperationExecutionListener() {
+        master.addOperationExecutionListener(new OperationExecutionListener<RacingEventService>() {
             @Override
-            public <T> void executed(OperationWithTransformationSupport<?, ? extends Operation<T>> operation) {
+            public <T> void executed(Operation<RacingEventService, T> operation) {
                 if (operation instanceof CreateTrackedRace) {
                     synchronized (notifier) {
                         notifier[0] = true;

java/com.sap.sailing.server/src/com/sap/sailing/server/RacingEventService.java

@@ -66,7 +66,6 @@ import com.sap.sailing.domain.tracking.TrackedRegattaRegistry;
 import com.sap.sailing.domain.tracking.TrackerManager;
 import com.sap.sailing.domain.tracking.WindStore;
 import com.sap.sailing.server.masterdata.DataImportLockWithProgress;
-import com.sap.sailing.server.replication.OperationExecutionListener;
 import com.sap.sailing.server.replication.Replicable;
 import com.sap.sse.common.TimePoint;
 import com.sap.sse.common.Util;

java/com.sap.sailing.server/src/com/sap/sailing/server/RacingEventServiceOperation.java

@@ -11,15 +11,9 @@ import com.sap.sailing.server.operationaltransformation.MoveLeaderboardColumnUp;
 import com.sap.sailing.server.operationaltransformation.RemoveLeaderboard;
 import com.sap.sailing.server.operationaltransformation.RemoveLeaderboardColumn;
 import com.sap.sailing.server.operationaltransformation.RenameLeaderboardColumn;
-import com.sap.sse.operationaltransformation.Operation;
+import com.sap.sailing.server.replication.OperationWithResult;
 
-public interface RacingEventServiceOperation<ResultType> extends Operation<ResultType>, Serializable {
+public interface RacingEventServiceOperation<ResultType> extends OperationWithResult<RacingEventService, ResultType>, Serializable {
-    /**
-     * Performs the actual operation, applying it to the <code>toState</code> service. The operation's result is
-     * returned.
-     */
-    ResultType internalApplyTo(RacingEventService toState) throws Exception;
-    
     /**
      * Assumes this is the "server" operation and transforms the client's <code>removeColumnFromLeaderboardClientOp</code> according to this
      * operation. The default implementation will probably pass on the untransformed client operation. However, if this

java/com.sap.sailing.server/src/com/sap/sailing/server/impl/RacingEventServiceImpl.java

@@ -296,7 +296,7 @@ public class RacingEventServiceImpl implements RacingEventServiceWithTestSupport
      */
     private final NamedReentrantReadWriteLock regattaTrackingCacheLock;
 
-    private final ConcurrentHashMap<OperationExecutionListener, OperationExecutionListener> operationExecutionListeners;
+    private final ConcurrentHashMap<OperationExecutionListener<RacingEventService>, OperationExecutionListener<RacingEventService>> operationExecutionListeners;
 
     /**
      * Keys are the toString() representation of the {@link RaceDefinition#getId() IDs} of races passed to
@@ -2084,7 +2084,7 @@ public class RacingEventServiceImpl implements RacingEventServiceWithTestSupport
      * this service known.
      */
     @Override
-    public <T> T apply(Operation<T> operation) {
+    public <T> T apply(Operation<RacingEventService> operation) {
         RacingEventServiceOperation<T> reso = (RacingEventServiceOperation<T>) operation;
         try {
             T result = reso.internalApplyTo(this);
@@ -2108,7 +2108,7 @@ public class RacingEventServiceImpl implements RacingEventServiceWithTestSupport
 
     @Override
     public <T> void replicate(RacingEventServiceOperation<T> operation) {
-        for (OperationExecutionListener listener : operationExecutionListeners.keySet()) {
+        for (OperationExecutionListener<RacingEventService> listener : operationExecutionListeners.keySet()) {
             try {
                 listener.executed(operation);
             } catch (Exception e) {
@@ -2120,12 +2120,12 @@ public class RacingEventServiceImpl implements RacingEventServiceWithTestSupport
     }
 
     @Override
-    public void addOperationExecutionListener(OperationExecutionListener listener) {
+    public void addOperationExecutionListener(OperationExecutionListener<RacingEventService> listener) {
         operationExecutionListeners.put(listener, listener);
     }
 
     @Override
-    public void removeOperationExecutionListener(OperationExecutionListener listener) {
+    public void removeOperationExecutionListener(OperationExecutionListener<RacingEventService> listener) {
         operationExecutionListeners.remove(listener);
     }
 

java/com.sap.sse.operationaltransformation.test/META-INF/MANIFEST.MF

@@ -5,5 +5,6 @@ Bundle-SymbolicName: com.sap.sse.operationaltransformation.test
 Bundle-Version: 1.0.0.qualifier
 Bundle-Vendor: SAP
 Bundle-RequiredExecutionEnvironment: JavaSE-1.8
-Require-Bundle: org.junit4;bundle-version="4.8.2"
+Require-Bundle: org.junit4;bundle-version="4.8.2",
+ com.sap.sse.replication
 Fragment-Host: com.sap.sse.operationaltransformation

java/com.sap.sse.operationaltransformation.test/src/com/sap/sse/operationaltransformation/test/OperationalTransformationTest.java

@@ -12,9 +12,9 @@ import org.junit.Test;
 import com.sap.sse.operationaltransformation.ClientServerOperationPair;
 import com.sap.sse.operationaltransformation.Operation;
 import com.sap.sse.operationaltransformation.Peer;
+import com.sap.sse.operationaltransformation.Peer.Role;
 import com.sap.sse.operationaltransformation.PeerImpl;
 import com.sap.sse.operationaltransformation.Transformer;
-import com.sap.sse.operationaltransformation.Peer.Role;
 import com.sap.sse.operationaltransformation.test.util.Base64;
 
 public class OperationalTransformationTest {

java/com.sap.sse.operationaltransformation/src/com/sap/sse/operationaltransformation/OperationWithTransformationSupport.java

@@ -1,7 +1,13 @@
 package com.sap.sse.operationaltransformation;
 
-
+/**
-
+ * Returns the same state object as was passed in. This is required for the Operational Transformation algorithm to work.
+ * 
+ * @author Axel Uhl (D043530)
+ *
+ * @param <S>
+ * @param <O>
+ */
 public interface OperationWithTransformationSupport<S, O extends OperationWithTransformationSupport<S, O>> extends Operation<S> {
     /**
      * Implements the specific transformation rule for the implementing subclass for the set of possible peer operations

java/com.sap.sse.operationaltransformation/src/com/sap/sse/operationaltransformation/OperationalTransformer.java

@@ -1,8 +1,15 @@
 package com.sap.sse.operationaltransformation;
 
-
+/**
+ * A default implementation for the {@link Transformer} interface, assuming operations implement the
+ * {@link OperationWithTransformationSupport} interface which standardizes how operations are being transformed.
+ * 
+ * @author Axel Uhl (D043530)
+ *
+ * @param <S>
+ * @param <O>
+ */
 public class OperationalTransformer<S, O extends OperationWithTransformationSupport<S, O>> implements Transformer<O> {
-
     @Override
     public ClientServerOperationPair<O> transform(O clientOp, O serverOp) {
         ClientServerOperationPair<O> result = new ClientServerOperationPair<O>(
@@ -10,5 +17,4 @@ public class OperationalTransformer<S, O extends OperationWithTransformationSupp
                         serverOp == null ? null : serverOp.transformServerOp(clientOp));
         return result;
     }
-
 }

java/com.sap.sse.replication/src/com/sap/sailing/server/replication/OperationExecutionListener.java

@@ -1,7 +1,14 @@
 package com.sap.sailing.server.replication;
 
-import com.sap.sse.operationaltransformation.Operation;
 
-public interface OperationExecutionListener {
+/**
-    <T> void executed(Operation<T> operation);
+ * Can be registered on a {@link Replicable} and will receive notifications about the execution of
+ * operations.
+ * 
+ * @author Axel Uhl (D043530)
+ *
+ * @param <S> the type of the state to which the operations are applied
+ */
+public interface OperationExecutionListener<S> {
+    <T> void executed(OperationWithResult<S, T> operation);
 }

java/com.sap.sse.replication/src/com/sap/sailing/server/replication/OperationWithResult.java

@@ -0,0 +1,20 @@
+package com.sap.sailing.server.replication;
+
+import com.sap.sse.operationaltransformation.Operation;
+
+/**
+ * An operational transformation {@link Operation} is expected to return the target state after applying the operation.
+ * This operation type offers the possibility to let an operation return a result value from its
+ * {@link #internalApplyTo} method while in the context of operational transformation the operation will still return
+ * the target state.
+ * 
+ * @author Axel Uhl (D043530)
+ *
+ */
+public interface OperationWithResult<S, R> extends Operation<S> {
+    /**
+     * Performs the actual operation, applying it to the <code>toState</code> service. The operation's result is
+     * returned.
+     */
+    R internalApplyTo(S toState) throws Exception;
+}

java/com.sap.sse.replication/src/com/sap/sailing/server/replication/Replicable.java

@@ -41,7 +41,7 @@ import com.sap.sse.operationaltransformation.OperationWithTransformationSupport;
  * @author Axel Uhl (D043530)
  *
  */
-public interface Replicable<S, O extends Operation<?>> extends WithID {
+public interface Replicable<S, O extends OperationWithResult<S, ?>> extends WithID {
     /**
      * The name of the property to use in the <code>properties</code> dictionary in a call to
      * {@link BundleContext#registerService(Class, Object, java.util.Dictionary)} when registering a {@link Replicable}.
@@ -61,11 +61,11 @@ public interface Replicable<S, O extends Operation<?>> extends WithID {
      * {@link OperationExecutionListener#executed(OperationWithTransformationSupport) notifies} all registered
      * operation execution listeners about the execution of the operation.
      */
-    <T> T apply(Operation<T> operation);
+    <T> T apply(OperationWithResult<S, T> operation);
 
-    void addOperationExecutionListener(OperationExecutionListener listener);
+    void addOperationExecutionListener(OperationExecutionListener<S> listener);
 
-    void removeOperationExecutionListener(OperationExecutionListener listener);
+    void removeOperationExecutionListener(OperationExecutionListener<S> listener);
 
     void clearReplicaState() throws MalformedURLException, IOException, InterruptedException;
 

java/com.sap.sse.replication/src/com/sap/sailing/server/replication/impl/Activator.java

@@ -111,7 +111,8 @@ public class Activator implements BundleActivator {
             logger.severe("Couldn't parse the replication port specification \""+exchangePortAsString+"\". Using default.");
         }
         replicationInstancesManager = new ReplicationInstancesManager();
-        ReplicationService serverReplicationMasterService = new ReplicationServiceImpl(exchangeName, exchangeHost, exchangePort, replicationInstancesManager);
+        ReplicationService serverReplicationMasterService = new ReplicationServiceImpl<>(
+                exchangeName, exchangeHost, exchangePort, replicationInstancesManager);
         bundleContext.registerService(ReplicationService.class, serverReplicationMasterService, null);
         logger.info("Registered replication service "+serverReplicationMasterService+" using exchange name "+exchangeName+" on host "+exchangeHost);
         checkIfAutomaticReplicationShouldStart(serverReplicationMasterService, exchangeName);

java/com.sap.sse.replication/src/com/sap/sailing/server/replication/impl/HasReplicable.java

@@ -1,7 +1,8 @@
 package com.sap.sailing.server.replication.impl;
 
+import com.sap.sailing.server.replication.OperationWithResult;
 import com.sap.sailing.server.replication.Replicable;
 
-public interface HasReplicable {
+public interface HasReplicable<S, O extends OperationWithResult<S, ?>> {
-    Replicable<?, ?> getReplicable();
+    Replicable<S, O> getReplicable();
 }

java/com.sap.sse.replication/src/com/sap/sailing/server/replication/impl/ReplicationServiceImpl.java

@@ -30,11 +30,11 @@ import com.rabbitmq.client.Channel;
 import com.rabbitmq.client.ConnectionFactory;
 import com.rabbitmq.client.QueueingConsumer;
 import com.sap.sailing.server.replication.OperationExecutionListener;
+import com.sap.sailing.server.replication.OperationWithResult;
 import com.sap.sailing.server.replication.Replicable;
 import com.sap.sailing.server.replication.ReplicationMasterDescriptor;
 import com.sap.sailing.server.replication.ReplicationService;
 import com.sap.sse.BuildVersion;
-import com.sap.sse.operationaltransformation.Operation;
 import com.sap.sse.operationaltransformation.OperationWithTransformationSupport;
 
 /**
@@ -52,14 +52,14 @@ import com.sap.sse.operationaltransformation.OperationWithTransformationSupport;
  * @author Frank Mittag, Axel Uhl (d043530)
  * 
  */
-public class ReplicationServiceImpl implements ReplicationService, OperationExecutionListener, HasReplicable {
+public class ReplicationServiceImpl<S, O extends OperationWithResult<S, ?>> implements ReplicationService, OperationExecutionListener<S>, HasReplicable<S, O> {
     private static final Logger logger = Logger.getLogger(ReplicationServiceImpl.class.getName());
     
     private final ReplicationInstancesManager replicationInstancesManager;
     
-    private final ServiceTracker<Replicable<?, ?>, Replicable<?, ?>> racingEventServiceTracker;
+    private final ServiceTracker<Replicable<S, O>, Replicable<S, O>> racingEventServiceTracker;
     
-    private final Replicable<?, ?> localService;
+    private final Replicable<S, O> localService;
     
     /**
      * <code>null</code>, if this instance is not currently replicating from some master; the master's descriptor otherwise
@@ -97,7 +97,7 @@ public class ReplicationServiceImpl implements ReplicationService, OperationExec
      */
     private final UUID serverUUID;
     
-    private Replicator replicator;
+    private Replicator<S, O> replicator;
     private Thread replicatorThread;
     
     /**
@@ -178,7 +178,7 @@ public class ReplicationServiceImpl implements ReplicationService, OperationExec
     
     private ReplicationServiceImpl(String exchangeName, String exchangeHost,
             int exchangePort, final ReplicationInstancesManager replicationInstancesManager,
-            Replicable<?, ?> localService, boolean createRacingEventServiceTracker) throws IOException {
+            Replicable<S, O> localService, boolean createRacingEventServiceTracker) throws IOException {
         timer = new Timer("ReplicationServiceImpl timer for delayed task sending");
         this.replicationInstancesManager = replicationInstancesManager;
         replicaUUIDs = new HashMap<ReplicationMasterDescriptor, String>();
@@ -206,12 +206,12 @@ public class ReplicationServiceImpl implements ReplicationService, OperationExec
      *            the name of the exchange to which replicas can bind
      */
     public ReplicationServiceImpl(String exchangeName, String exchangeHost,
-            final ReplicationInstancesManager replicationInstancesManager, Replicable<?, ?> localService) throws IOException {
+            final ReplicationInstancesManager replicationInstancesManager, Replicable<S, O> localService) throws IOException {
         this(exchangeName, exchangeHost, 0, replicationInstancesManager, localService, /* create RacingEventServiceTracker */ false);
     }
     
-    protected ServiceTracker<Replicable<?, ?>, Replicable<?, ?>> getRacingEventServiceTracker() {
+    protected ServiceTracker<Replicable<S, O>, Replicable<S, O>> getRacingEventServiceTracker() {
-        return new ServiceTracker<Replicable<?, ?>, Replicable<?, ?>>(
+        return new ServiceTracker<Replicable<S, O>, Replicable<S, O>>(
                 Activator.getDefaultContext(), Replicable.class.getName(), null);
     }
     
@@ -242,8 +242,8 @@ public class ReplicationServiceImpl implements ReplicationService, OperationExec
     }
 
     @Override
-    public Replicable<?, ?> getReplicable() {
+    public Replicable<S, O> getReplicable() {
-        Replicable<?, ?> result;
+        Replicable<S, O> result;
         if (localService != null) {
             result = localService;
         } else {
@@ -293,7 +293,7 @@ public class ReplicationServiceImpl implements ReplicationService, OperationExec
      * Schedules a single operation for broadcast. The operation is added to {@link #outboundBuffer}, and if not already scheduled,
      * a {@link #timer} is created and scheduled to send in {@link #TRANSMISSION_DELAY_MILLIS} milliseconds.
      */
-    private void broadcastOperation(Operation<?> operation) throws IOException {
+    private void broadcastOperation(OperationWithResult<S, ?> operation) throws IOException {
         // need to write the operations one by one, making sure the ObjectOutputStream always writes
         // identical objects again if required because they may have changed state in between
         ByteArrayOutputStream bos = new ByteArrayOutputStream();
@@ -426,7 +426,7 @@ public class ReplicationServiceImpl implements ReplicationService, OperationExec
         logger.info("Connection to exchange successful.");
         URL initialLoadURL = master.getInitialLoadURL();
         logger.info("Initial load URL is "+initialLoadURL);
-        replicator = new Replicator(master, this, /* startSuspended */ true, consumer);
+        replicator = new Replicator<S, O>(master, this, /* startSuspended */ true, consumer);
         // start receiving messages already now, but start in suspended mode
         replicatorThread = new Thread(replicator, "Replicator receiving from "+master.getMessagingHostname()+"/"+master.getExchangeName());
         final Replicable<?, ?> replicable = getReplicable();
@@ -497,7 +497,7 @@ public class ReplicationServiceImpl implements ReplicationService, OperationExec
      * replicas by publishing it to the fan-out exchange.
      */
     @Override
-    public <T> void executed(Operation<T> operation) {
+    public <T> void executed(OperationWithResult<S, T> operation) {
         try {
             broadcastOperation(operation);
         } catch (Exception e) {

java/com.sap.sse.replication/src/com/sap/sailing/server/replication/impl/Replicator.java

@@ -21,9 +21,8 @@ import com.rabbitmq.client.ConsumerCancelledException;
 import com.rabbitmq.client.QueueingConsumer;
 import com.rabbitmq.client.QueueingConsumer.Delivery;
 import com.rabbitmq.client.ShutdownSignalException;
+import com.sap.sailing.server.replication.OperationWithResult;
 import com.sap.sailing.server.replication.ReplicationMasterDescriptor;
-import com.sap.sse.operationaltransformation.Operation;
-import com.sap.sse.operationaltransformation.OperationWithTransformationSupport;
 
 /**
  * Receives {@link RacingEventServiceOperation}s through JMS and
@@ -38,15 +37,15 @@ import com.sap.sse.operationaltransformation.OperationWithTransformationSupport;
  * @author Axel Uhl (d043530)
  * 
  */
-public class Replicator implements Runnable {
+public class Replicator<S, O extends OperationWithResult<S, ?>> implements Runnable {
     private final static Logger logger = Logger.getLogger(Replicator.class.getName());
     
     private static final long CHECK_INTERVAL_MILLIS = 2000; // how long (milliseconds) to pause before checking connection again
     private static final int CHECK_COUNT = 150; // how long to check, value is CHECK_INTERVAL second steps
     
     private final ReplicationMasterDescriptor master;
-    private final HasReplicable replicableProvider;
+    private final HasReplicable<S, O> replicableProvider;
-    private final List<Operation<?>> queue;
+    private final List<OperationWithResult<S, ?>> queue;
     
     private QueueingConsumer consumer;
     
@@ -89,8 +88,8 @@ public class Replicator implements Runnable {
      * @param consumer
      *            the RabbitMQ consumer from which to load messages
      */
-    public Replicator(ReplicationMasterDescriptor master, HasReplicable racingEventServiceTracker, boolean startSuspended, QueueingConsumer consumer) {
+    public Replicator(ReplicationMasterDescriptor master, HasReplicable<S, O> racingEventServiceTracker, boolean startSuspended, QueueingConsumer consumer) {
-        this.queue = new ArrayList<Operation<?>>();
+        this.queue = new ArrayList<OperationWithResult<S, ?>>();
         this.master = master;
         this.replicableProvider = racingEventServiceTracker;
         this.suspended = startSuspended;
@@ -147,7 +146,8 @@ public class Replicator implements Runnable {
                         byte[] serializedOperation = (byte[]) ois.readObject();
                         ObjectInputStream operationOIS = replicableProvider.getReplicable()
                                 .createObjectInputStreamResolvingAgainstCache(new ByteArrayInputStream(serializedOperation));
-                        OperationWithTransformationSupport<?, ?> operation = (OperationWithTransformationSupport<?, ?>) operationOIS.readObject();
+                        @SuppressWarnings("unchecked")
+                        OperationWithResult<S, ?> operation = (OperationWithResult<S, ?>) operationOIS.readObject();
                         operationCount++;
                         operationsInMessage++;
                         if (operationCount % 10000l == 0) {
@@ -222,7 +222,7 @@ public class Replicator implements Runnable {
      * If the replicator is currently {@link #suspended}, the <code>operation</code> is queued, otherwise immediately applied to
      * the receiving replica.
      */
-    private synchronized void applyOrQueue(OperationWithTransformationSupport<?, ?> operation) {
+    private synchronized void applyOrQueue(OperationWithResult<S, ?> operation) {
         if (suspended) {
             queue(operation);
         } else {
@@ -230,7 +230,7 @@ public class Replicator implements Runnable {
         }
     }
 
-    private synchronized void apply(final Operation<?> operation) {
+    private synchronized void apply(final OperationWithResult<S, ?> operation) {
         Runnable runnable = new Runnable() {
             @Override
             public void run() {
@@ -244,7 +244,7 @@ public class Replicator implements Runnable {
         }
     }
     
-    private synchronized void queue(Operation<?> operation) {
+    private synchronized void queue(OperationWithResult<S, ?> operation) {
         if (queue.isEmpty()) {
             notifyAll();
         }
@@ -262,8 +262,8 @@ public class Replicator implements Runnable {
     }
     
     private synchronized void applyQueue() {
-        for (Iterator<Operation<?>> i=queue.iterator(); i.hasNext(); ) {
+        for (Iterator<OperationWithResult<S, ?>> i=queue.iterator(); i.hasNext(); ) {
-            Operation<?> operation = i.next();
+            OperationWithResult<S, ?> operation = i.next();
             i.remove();
             try {
                 apply(operation);