feat: ensure consistency of called CrrTransmitCallback instances of a CompositeCrrTransmitCallback process

oldratlee · oldratlee · commit 4ba3dd663f1c · 2022-12-14T00:49:45.000+08:00
diff --git a/ttl-core/src/main/java/com/alibaba/crr/composite/CompositeCrrTransmit.java b/ttl-core/src/main/java/com/alibaba/crr/composite/CompositeCrrTransmit.java
@@ -1,7 +1,6 @@
 package com.alibaba.crr.composite;
 
 import com.alibaba.crr.CrrTransmit;
-import com.alibaba.crr.CrrTransmitCallback;
 import edu.umd.cs.findbugs.annotations.NonNull;
 
 import java.util.HashMap;
@@ -24,14 +23,14 @@
  *
  * @author Jerry Lee (oldratlee at gmail dot com)
  */
-public class CompositeCrrTransmit implements CrrTransmit<Capture, Backup> {
+public final class CompositeCrrTransmit implements CrrTransmit<Capture, Backup> {
     private static final Logger logger = Logger.getLogger(CompositeCrrTransmit.class.getName());
 
     private final Set<CrrTransmit<Object, Object>> registeredCrrTransmitSet = new CopyOnWriteArraySet<>();
 
-    private final CrrTransmitCallback callback;
+    private final CompositeCrrTransmitCallback callback;
 
-    public CompositeCrrTransmit(CrrTransmitCallback callback) {
+    public CompositeCrrTransmit(CompositeCrrTransmitCallback callback) {
         this.callback = callback;
     }
 
@@ -53,7 +52,7 @@ public Capture capture() {
                 }
             }
         }
-        return new Snapshot(crrTransmit2Value);
+        return new Snapshot(crrTransmit2Value, null);
     }
 
     /**
@@ -66,9 +65,9 @@ public Capture capture() {
      */
     @NonNull
     public Backup replay(@NonNull Capture captured) {
-        callback.beforeReplay();
-        final Snapshot capturedSnapshot = (Snapshot) captured;
+        final Object data = callback.beforeReplay();
 
+        final Snapshot capturedSnapshot = (Snapshot) captured;
         final HashMap<CrrTransmit<Object, Object>, Object> crrTransmit2Value = new HashMap<>(capturedSnapshot.crrTransmit2Value.size());
         for (Map.Entry<CrrTransmit<Object, Object>, Object> entry : capturedSnapshot.crrTransmit2Value.entrySet()) {
             CrrTransmit<Object, Object> crrTransmit = entry.getKey();
@@ -82,8 +81,9 @@ public Backup replay(@NonNull Capture captured) {
                 }
             }
         }
-        callback.afterReplay();
-        return new Snapshot(crrTransmit2Value);
+
+        final Object afterData = callback.afterReplay(data);
+        return new Snapshot(crrTransmit2Value, afterData);
     }
 
     /**
@@ -103,7 +103,8 @@ public Backup replay(@NonNull Capture captured) {
      */
     @NonNull
     public Backup clear() {
-        callback.beforeReplay();
+        final Object data = callback.beforeReplay();
+
         final HashMap<CrrTransmit<Object, Object>, Object> crrTransmit2Value = new HashMap<>(registeredCrrTransmitSet.size());
         for (CrrTransmit<Object, Object> crrTransmit : registeredCrrTransmitSet) {
             try {
@@ -115,8 +116,9 @@ public Backup clear() {
                 }
             }
         }
-        callback.afterReplay();
-        return new Snapshot(crrTransmit2Value);
+
+        final Object afterData = callback.afterReplay(data);
+        return new Snapshot(crrTransmit2Value, afterData);
     }
 
     /**
@@ -127,8 +129,10 @@ public Backup clear() {
      * @see #clear()
      */
     public void restore(@NonNull Backup backup) {
-        callback.beforeRestore();
-        for (Map.Entry<CrrTransmit<Object, Object>, Object> entry : ((Snapshot) backup).crrTransmit2Value.entrySet()) {
+        final Snapshot snapshot = (Snapshot) backup;
+        final Object data = callback.beforeRestore(snapshot.data);
+
+        for (Map.Entry<CrrTransmit<Object, Object>, Object> entry : snapshot.crrTransmit2Value.entrySet()) {
             CrrTransmit<Object, Object> crrTransmit = entry.getKey();
             try {
                 Object transmitBackup = entry.getValue();
@@ -140,14 +144,17 @@ public void restore(@NonNull Backup backup) {
                 }
             }
         }
-        callback.afterRestore();
+
+        callback.afterRestore(data);
     }
 
     private static class Snapshot implements Capture, Backup {
         final HashMap<CrrTransmit<Object, Object>, Object> crrTransmit2Value;
+        final Object data;
 
-        Snapshot(HashMap<CrrTransmit<Object, Object>, Object> crrTransmit2Value) {
+        Snapshot(HashMap<CrrTransmit<Object, Object>, Object> crrTransmit2Value, Object data) {
             this.crrTransmit2Value = crrTransmit2Value;
+            this.data = data;
         }
     }
 
diff --git a/ttl-core/src/main/java/com/alibaba/crr/composite/CompositeCrrTransmitCallback.java b/ttl-core/src/main/java/com/alibaba/crr/composite/CompositeCrrTransmitCallback.java
@@ -3,6 +3,7 @@
 import com.alibaba.crr.CrrTransmitCallback;
 import edu.umd.cs.findbugs.annotations.NonNull;
 
+import java.util.HashSet;
 import java.util.Set;
 import java.util.concurrent.CopyOnWriteArraySet;
 import java.util.logging.Level;
@@ -14,62 +15,68 @@
  * @author Jerry Lee (oldratlee at gmail dot com)
  * @see CrrTransmitCallback
  */
-public class CompositeCrrTransmitCallback implements CrrTransmitCallback {
+public final class CompositeCrrTransmitCallback {
     private static final Logger logger = Logger.getLogger(CompositeCrrTransmitCallback.class.getName());
 
     private final Set<CrrTransmitCallback> registeredCrrTransmitCallbackSet = new CopyOnWriteArraySet<>();
 
-    @Override
-    public void beforeReplay() {
-        for (CrrTransmitCallback callback : registeredCrrTransmitCallbackSet) {
+    Object beforeReplay() {
+        Set<CrrTransmitCallback> callbacks = new HashSet<>(registeredCrrTransmitCallbackSet);
+        for (CrrTransmitCallback cb : callbacks) {
             try {
-                callback.beforeReplay();
+                cb.beforeReplay();
             } catch (Throwable t) {
                 if (logger.isLoggable(Level.WARNING)) {
-                    logger.log(Level.WARNING, "exception when beforeReplay for crrTransmitCallback " + callback +
-                            "(class " + callback.getClass().getName() + "), just ignored; cause: " + t, t);
+                    logger.log(Level.WARNING, "exception when beforeReplay for crrTransmitCallback " + cb +
+                            "(class " + cb.getClass().getName() + "), just ignored; cause: " + t, t);
                 }
             }
         }
+        return callbacks;
     }
 
-    @Override
-    public void afterReplay() {
-        for (CrrTransmitCallback callback : registeredCrrTransmitCallbackSet) {
+    Object afterReplay(Object data) {
+        @SuppressWarnings("unchecked")
+        Set<CrrTransmitCallback> callbacks = (Set<CrrTransmitCallback>) data;
+        for (CrrTransmitCallback cb : callbacks) {
             try {
-                callback.afterReplay();
+                cb.afterReplay();
             } catch (Throwable t) {
                 if (logger.isLoggable(Level.WARNING)) {
-                    logger.log(Level.WARNING, "exception when afterReplay for crrTransmitCallback " + callback +
-                            "(class " + callback.getClass().getName() + "), just ignored; cause: " + t, t);
+                    logger.log(Level.WARNING, "exception when afterReplay for crrTransmitCallback " + cb +
+                            "(class " + cb.getClass().getName() + "), just ignored; cause: " + t, t);
                 }
             }
         }
+        return data;
     }
 
-    @Override
-    public void beforeRestore() {
-        for (CrrTransmitCallback callback : registeredCrrTransmitCallbackSet) {
+    Object beforeRestore(Object data) {
+        @SuppressWarnings("unchecked")
+        Set<CrrTransmitCallback> callbacks = (Set<CrrTransmitCallback>) data;
+        for (CrrTransmitCallback cb : callbacks) {
             try {
-                callback.beforeRestore();
+                cb.beforeRestore();
             } catch (Throwable t) {
                 if (logger.isLoggable(Level.WARNING)) {
-                    logger.log(Level.WARNING, "exception when beforeRestore for crrTransmitCallback " + callback +
-                            "(class " + callback.getClass().getName() + "), just ignored; cause: " + t, t);
+                    logger.log(Level.WARNING, "exception when beforeRestore for crrTransmitCallback " + cb +
+                            "(class " + cb.getClass().getName() + "), just ignored; cause: " + t, t);
                 }
             }
         }
+        return data;
     }
 
-    @Override
-    public void afterRestore() {
-        for (CrrTransmitCallback callback : registeredCrrTransmitCallbackSet) {
+    void afterRestore(Object data) {
+        @SuppressWarnings("unchecked")
+        Set<CrrTransmitCallback> callbacks = (Set<CrrTransmitCallback>) data;
+        for (CrrTransmitCallback cb : callbacks) {
             try {
-                callback.afterRestore();
+                cb.afterRestore();
             } catch (Throwable t) {
                 if (logger.isLoggable(Level.WARNING)) {
-                    logger.log(Level.WARNING, "exception when afterRestore for crrTransmitCallback " + callback +
-                            "(class " + callback.getClass().getName() + "), just ignored; cause: " + t, t);
+                    logger.log(Level.WARNING, "exception when afterRestore for crrTransmitCallback " + cb +
+                            "(class " + cb.getClass().getName() + "), just ignored; cause: " + t, t);
                 }
             }
         }

Original file line number	Diff line number	Diff line change
`@@ -1,7 +1,6 @@`
`1`	`1`	`package com.alibaba.crr.composite;`
`2`	`2`
`3`	`3`	`import com.alibaba.crr.CrrTransmit;`
`4`		`-import com.alibaba.crr.CrrTransmitCallback;`
`5`	`4`	`import edu.umd.cs.findbugs.annotations.NonNull;`
`6`	`5`
`7`	`6`	`import java.util.HashMap;`
`@@ -24,14 +23,14 @@`
`24`	`23`	`*`
`25`	`24`	`* @author Jerry Lee (oldratlee at gmail dot com)`
`26`	`25`	`*/`
`27`		`-public class CompositeCrrTransmit implements CrrTransmit<Capture, Backup> {`
	`26`	`+public final class CompositeCrrTransmit implements CrrTransmit<Capture, Backup> {`
`28`	`27`	`private static final Logger logger = Logger.getLogger(CompositeCrrTransmit.class.getName());`
`29`	`28`
`30`	`29`	`private final Set<CrrTransmit<Object, Object>> registeredCrrTransmitSet = new CopyOnWriteArraySet<>();`
`31`	`30`
`32`		`- private final CrrTransmitCallback callback;`
	`31`	`+ private final CompositeCrrTransmitCallback callback;`
`33`	`32`
`34`		`- public CompositeCrrTransmit(CrrTransmitCallback callback) {`
	`33`	`+ public CompositeCrrTransmit(CompositeCrrTransmitCallback callback) {`
`35`	`34`	`this.callback = callback;`
`36`	`35`	`}`
`37`	`36`
`@@ -53,7 +52,7 @@ public Capture capture() {`
`53`	`52`	`}`
`54`	`53`	`}`
`55`	`54`	`}`
`56`		`- return new Snapshot(crrTransmit2Value);`
	`55`	`+ return new Snapshot(crrTransmit2Value, null);`
`57`	`56`	`}`
`58`	`57`
`59`	`58`	`/**`
`@@ -66,9 +65,9 @@ public Capture capture() {`
`66`	`65`	`*/`
`67`	`66`	`@NonNull`
`68`	`67`	`public Backup replay(@NonNull Capture captured) {`
`69`		`- callback.beforeReplay();`
`70`		`- final Snapshot capturedSnapshot = (Snapshot) captured;`
	`68`	`+ final Object data = callback.beforeReplay();`
`71`	`69`
	`70`	`+ final Snapshot capturedSnapshot = (Snapshot) captured;`
`72`	`71`	`final HashMap<CrrTransmit<Object, Object>, Object> crrTransmit2Value = new HashMap<>(capturedSnapshot.crrTransmit2Value.size());`
`73`	`72`	`for (Map.Entry<CrrTransmit<Object, Object>, Object> entry : capturedSnapshot.crrTransmit2Value.entrySet()) {`
`74`	`73`	`CrrTransmit<Object, Object> crrTransmit = entry.getKey();`
`@@ -82,8 +81,9 @@ public Backup replay(@NonNull Capture captured) {`
`82`	`81`	`}`
`83`	`82`	`}`
`84`	`83`	`}`
`85`		`- callback.afterReplay();`
`86`		`- return new Snapshot(crrTransmit2Value);`
	`84`	`+`
	`85`	`+ final Object afterData = callback.afterReplay(data);`
	`86`	`+ return new Snapshot(crrTransmit2Value, afterData);`
`87`	`87`	`}`
`88`	`88`
`89`	`89`	`/**`
`@@ -103,7 +103,8 @@ public Backup replay(@NonNull Capture captured) {`
`103`	`103`	`*/`
`104`	`104`	`@NonNull`
`105`	`105`	`public Backup clear() {`
`106`		`- callback.beforeReplay();`
	`106`	`+ final Object data = callback.beforeReplay();`
	`107`	`+`
`107`	`108`	`final HashMap<CrrTransmit<Object, Object>, Object> crrTransmit2Value = new HashMap<>(registeredCrrTransmitSet.size());`
`108`	`109`	`for (CrrTransmit<Object, Object> crrTransmit : registeredCrrTransmitSet) {`
`109`	`110`	`try {`
`@@ -115,8 +116,9 @@ public Backup clear() {`
`115`	`116`	`}`
`116`	`117`	`}`
`117`	`118`	`}`
`118`		`- callback.afterReplay();`
`119`		`- return new Snapshot(crrTransmit2Value);`
	`119`	`+`
	`120`	`+ final Object afterData = callback.afterReplay(data);`
	`121`	`+ return new Snapshot(crrTransmit2Value, afterData);`
`120`	`122`	`}`
`121`	`123`
`122`	`124`	`/**`
`@@ -127,8 +129,10 @@ public Backup clear() {`
`127`	`129`	`* @see #clear()`
`128`	`130`	`*/`
`129`	`131`	`public void restore(@NonNull Backup backup) {`
`130`		`- callback.beforeRestore();`
`131`		`- for (Map.Entry<CrrTransmit<Object, Object>, Object> entry : ((Snapshot) backup).crrTransmit2Value.entrySet()) {`
	`132`	`+ final Snapshot snapshot = (Snapshot) backup;`
	`133`	`+ final Object data = callback.beforeRestore(snapshot.data);`
	`134`	`+`
	`135`	`+ for (Map.Entry<CrrTransmit<Object, Object>, Object> entry : snapshot.crrTransmit2Value.entrySet()) {`
`132`	`136`	`CrrTransmit<Object, Object> crrTransmit = entry.getKey();`
`133`	`137`	`try {`
`134`	`138`	`Object transmitBackup = entry.getValue();`
`@@ -140,14 +144,17 @@ public void restore(@NonNull Backup backup) {`
`140`	`144`	`}`
`141`	`145`	`}`
`142`	`146`	`}`
`143`		`- callback.afterRestore();`
	`147`	`+`
	`148`	`+ callback.afterRestore(data);`
`144`	`149`	`}`
`145`	`150`
`146`	`151`	`private static class Snapshot implements Capture, Backup {`
`147`	`152`	`final HashMap<CrrTransmit<Object, Object>, Object> crrTransmit2Value;`
	`153`	`+ final Object data;`
`148`	`154`
`149`		`- Snapshot(HashMap<CrrTransmit<Object, Object>, Object> crrTransmit2Value) {`
	`155`	`+ Snapshot(HashMap<CrrTransmit<Object, Object>, Object> crrTransmit2Value, Object data) {`
`150`	`156`	`this.crrTransmit2Value = crrTransmit2Value;`
	`157`	`+ this.data = data;`
`151`	`158`	`}`
`152`	`159`	`}`
`153`	`160`