feat: ensure consistency of called CrrTransmitCallback instances of…

… a `CompositeCrrTransmitCallback` process
alibaba · Dec 13, 2022 · 4ba3dd6 · 4ba3dd6
1 parent d4c0346
commit 4ba3dd6
Show file tree

Hide file tree

Showing 2 changed files with 55 additions and 41 deletions.
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);
                 }
             }
         }