[Issue #768] fix issue in update HTTP subscriber

eventmesh-common/src/main/java/org/apache/eventmesh/common/protocol/SubscriptionItem.java

@@ -17,7 +17,11 @@
 
 package org.apache.eventmesh.common.protocol;
 
-public class SubscriptionItem {
+import java.io.Serializable;
+
+import com.google.common.base.Objects;
+
+public class SubscriptionItem implements Serializable {
 
     private String topic;
 
@@ -66,6 +70,23 @@ public String toString() {
                 + ", type=" + type
                 + '}';
     }
+
+    @Override
+    public boolean equals(Object o) {
+        if (this == o) {
+            return true;
+        }
+        if (o == null || getClass() != o.getClass()) {
+            return false;
+        }
+        SubscriptionItem that = (SubscriptionItem) o;
+        return Objects.equal(topic, that.topic) && mode == that.mode && type == that.type;
+    }
+
+    @Override
+    public int hashCode() {
+        return Objects.hashCode(topic, mode, type);
+    }
 }
 
 

eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/consumergroup/ConsumerGroupConf.java

@@ -17,12 +17,13 @@
 
 package org.apache.eventmesh.runtime.core.consumergroup;
 
+import java.io.Serializable;
 import java.util.Map;
 import java.util.Objects;
 
 import com.google.common.collect.Maps;
 
-public class ConsumerGroupConf {
+public class ConsumerGroupConf implements Serializable {
     //eg . 5013-1A0
     private String consumerGroup;
 

eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/consumergroup/ConsumerGroupTopicConf.java

@@ -19,6 +19,7 @@
 
 import org.apache.eventmesh.common.protocol.SubscriptionItem;
 
+import java.io.Serializable;
 import java.util.List;
 import java.util.Map;
 import java.util.Objects;
@@ -30,7 +31,7 @@
 import com.google.common.collect.Maps;
 import com.google.common.collect.Sets;
 
-public class ConsumerGroupTopicConf {
+public class ConsumerGroupTopicConf implements Serializable {
 
     public static Logger logger = LoggerFactory.getLogger(ConsumerGroupTopicConf.class);
 

eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/consumer/ConsumerManager.java

@@ -24,6 +24,7 @@
 import org.apache.eventmesh.runtime.core.consumergroup.event.ConsumerGroupStateEvent;
 import org.apache.eventmesh.runtime.core.consumergroup.event.ConsumerGroupTopicConfChangeEvent;
 import org.apache.eventmesh.runtime.core.protocol.http.processor.inf.Client;
+import org.apache.eventmesh.runtime.util.EventMeshUtil;
 
 import org.apache.commons.lang3.StringUtils;
 
@@ -192,7 +193,7 @@ public void notifyConsumerManager(String consumerGroup,
             ConsumerGroupStateEvent notification = new ConsumerGroupStateEvent();
             notification.action = ConsumerGroupStateEvent.ConsumerGroupStateAction.NEW;
             notification.consumerGroup = consumerGroup;
-            notification.consumerGroupConfig = latestConsumerGroupConfig;
+            notification.consumerGroupConfig = EventMeshUtil.cloneObject(latestConsumerGroupConfig);
             eventMeshHTTPServer.getEventBus().post(notification);
             return;
         }
@@ -201,7 +202,7 @@ public void notifyConsumerManager(String consumerGroup,
             ConsumerGroupStateEvent notification = new ConsumerGroupStateEvent();
             notification.action = ConsumerGroupStateEvent.ConsumerGroupStateAction.CHANGE;
             notification.consumerGroup = consumerGroup;
-            notification.consumerGroupConfig = latestConsumerGroupConfig;
+            notification.consumerGroupConfig = EventMeshUtil.cloneObject(latestConsumerGroupConfig);
             eventMeshHTTPServer.getEventBus().post(notification);
             return;
         }

eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/util/EventMeshUtil.java

@@ -26,6 +26,11 @@
 
 import org.apache.commons.lang3.StringUtils;
 
+import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.io.ObjectInputStream;
+import java.io.ObjectOutputStream;
 import java.net.Inet6Address;
 import java.net.InetAddress;
 import java.net.NetworkInterface;
@@ -344,4 +349,21 @@ public static void printState(ThreadPoolExecutor scheduledExecutorService) {
                 .getPoolSize(), scheduledExecutorService.getActiveCount(), scheduledExecutorService
                 .getCompletedTaskCount());
     }
+
+    /**
+     * Perform deep clone of the given object using serialization
+     * @param object
+     * @return cloned object
+     * @throws IOException
+     * @throws ClassNotFoundException
+     */
+    public static <T> T cloneObject(T object) throws IOException, ClassNotFoundException {
+        ByteArrayOutputStream byOut = new ByteArrayOutputStream();
+        ObjectOutputStream outputStream = new ObjectOutputStream(byOut);
+        outputStream.writeObject(object);
+
+        ByteArrayInputStream byIn = new ByteArrayInputStream(byOut.toByteArray());
+        ObjectInputStream inputStream = new ObjectInputStream(byIn);
+        return (T) inputStream.readObject();
+    }
 }