bug5741: don't serialize replication operations into separate byte[] anymore

java/com.sap.sse.replication.interfaces/src/com/sap/sse/replication/OperationWithResult.java

@@ -1,5 +1,6 @@
 package com.sap.sse.replication;
 
+import java.io.ObjectOutputStream;
 import java.io.Serializable;
 
 import com.sap.sse.operationaltransformation.Operation;
@@ -8,7 +9,21 @@ 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.
+ * the target state.<p>
+ * 
+ * Operations that are instance of a class implementing this interface inherit the serializability declared by this
+ * interface. Multiple operation instances may be serialized into the same {@link ObjectOutputStream} (see also
+ * bug 5741). Keep this in mind when choosing how to represent the operation's internals. For example, it is not a
+ * good idea to have the operation reference a mutable object with the intention to transport a new "to-be" state
+ * this way. If multiple such state changes end up in the same stream of operations, the mutable object will be
+ * serialized to the stream only once, in one (probably the first, thus oldest) state, and all subsequent operations
+ * referencing that same object will serialize only a handle referencing the object state already written. With this,
+ * you may lose state changes. Hence, make sure to either clone such objects before storing the clone in the operation,
+ * or---even better---use specialized operations describing the specific state change to apply. For example, instead
+ * of serializing an object with all its attributes, rather implement an operation per attribute that updates only
+ * that attribute's value. It may be more tedious, but it keeps your serialized operations small, is therefore more
+ * bandwidth-efficient in the replication architecture and less prone to accidentally dropping state changes by
+ * representing the same object in the stream only once.
  * 
  * @param <S>
  *            type of state to which the operation can be applied

java/com.sap.sse.replication.interfaces/src/com/sap/sse/replication/Replicable.java

@@ -165,6 +165,12 @@ extends OperationsToMasterSender<S, O>, Replicator<S, O> {
      */
     void clearReplicaState() throws MalformedURLException, IOException, InterruptedException;
 
+    /**
+     * Produces an object input stream that can choose to resolve objects against a cache so that duplicate instances
+     * are avoided.
+     */
+    ObjectInputStream createObjectInputStreamResolvingAgainstCache(InputStream is, Map<String, Class<?>> classLoaderCache) throws IOException;
+
     /**
      * Dual, reading operation for {@link #serializeForInitialReplication(OutputStream)}. In other words, when this
      * operation returns, this service instance is in a state "equivalent" to that of the service instance that produced
@@ -187,6 +193,30 @@ extends OperationsToMasterSender<S, O>, Replicator<S, O> {
      */
     void serializeForInitialReplication(OutputStream os) throws IOException;
 
+    /**
+     * The class loader to use for de-serializing objects. By default, this object's class's class loader is used.
+     */
+    default ClassLoader getDeserializationClassLoader() {
+        return getClass().getClassLoader();
+    }
+    
+    /**
+     * Implementation of {@link #readOperation(InputStream, Map)}, using the {@link ObjectInputStream} created by
+     * {@link #createObjectInputStreamResolvingAgainstCache(InputStream, Map)}. Before actually reading an operation
+     * object, the current thread's context class loader is set to the {@link #getDeserializationClassLoader() class
+     * loader for de-serialization} and restored to its previous value in the {@code finally} clause.
+     */
+    @SuppressWarnings("unchecked")
+    default O readOperationFromObjectInputStream(ObjectInputStream ois) throws ClassNotFoundException, IOException {
+        ClassLoader oldContextClassloader = Thread.currentThread().getContextClassLoader();
+        Thread.currentThread().setContextClassLoader(getDeserializationClassLoader());
+        try {
+            return (O) ois.readObject();
+        } finally {
+            Thread.currentThread().setContextClassLoader(oldContextClassloader);
+        }
+    }
+
     /**
      * From an input stream, reads an operation that can be {@link #apply(OperationWithResult) applied} to this object.
      * Separating reading and applying gives clients an opportunity to queue operations, e.g., in order to wait until

java/com.sap.sse.replication.interfaces/src/com/sap/sse/replication/ReplicableWithObjectInputStream.java

@@ -32,12 +32,6 @@ public interface ReplicableWithObjectInputStream<S, O extends OperationWithResul
     
     static final AtomicInteger operationCounter = new AtomicInteger(0);
     
-    /**
-     * Produces an object input stream that can choose to resolve objects against a cache so that duplicate instances
-     * are avoided.
-     */
-    ObjectInputStream createObjectInputStreamResolvingAgainstCache(InputStream is, Map<String, Class<?>> classLoaderCache) throws IOException;
-
     /**
      * Implementation of {@link #initiallyFillFrom(InputStream)} which receives an {@link ObjectInputStream} instead of
      * an {@link InputStream}. The {@link ObjectInputStream} is expected to have been produced by
@@ -50,15 +44,6 @@ public interface ReplicableWithObjectInputStream<S, O extends OperationWithResul
      */
     void serializeForInitialReplicationInternal(ObjectOutputStream objectOutputStream) throws IOException;
 
-    /**
-     * Implementation of {@link #readOperation(InputStream, Map)}, using the {@link ObjectInputStream} created by
-     * {@link #createObjectInputStreamResolvingAgainstCache(InputStream, Map)}.
-     */
-    @SuppressWarnings("unchecked")
-    default O readOperationInternal(ObjectInputStream ois) throws ClassNotFoundException, IOException {
-        return (O) ois.readObject();
-    }
-
     @Override
     default void initiallyFillFrom(InputStream is) throws IOException, ClassNotFoundException, InterruptedException {
         assert !isCurrentlyFillingFromInitialLoadOrApplyingOperationReceivedFromMaster(); // no nested receiving of initial load
@@ -77,13 +62,6 @@ public interface ReplicableWithObjectInputStream<S, O extends OperationWithResul
         }
     }
     
-    /**
-     * The class loader to use for de-serializing objects. By default, this object's class's class loader is used.
-     */
-    default ClassLoader getDeserializationClassLoader() {
-        return getClass().getClassLoader();
-    }
-    
     /**
      * Wraps <code>os</code> by an {@link ObjectOutputStream} and invokes
      * {@link #serializeForInitialReplicationInternal(ObjectOutputStream)}.
@@ -97,13 +75,7 @@ public interface ReplicableWithObjectInputStream<S, O extends OperationWithResul
     
     @Override
     default O readOperation(InputStream inputStream, Map<String, Class<?>> classLoaderCache) throws IOException, ClassNotFoundException {
-        ClassLoader oldContextClassloader = Thread.currentThread().getContextClassLoader();
-        Thread.currentThread().setContextClassLoader(getDeserializationClassLoader());
-        try {
-            return readOperationInternal(createObjectInputStreamResolvingAgainstCache(inputStream, classLoaderCache));
-        } finally {
-            Thread.currentThread().setContextClassLoader(oldContextClassloader);
-        }
+        return readOperationFromObjectInputStream(createObjectInputStreamResolvingAgainstCache(inputStream, classLoaderCache));
     }
 
     @Override