[Dubbo-4525] same as #4533, fix ClusterUtils.mergeUrl (#4540)

apache · Jul 12, 2019 · 0ea3e28 · 0ea3e28
1 parent 7e041b2
commit 0ea3e28
Show file tree

Hide file tree

Showing 2 changed files with 14 additions and 10 deletions.
diff --git a/dubbo-cluster/src/main/java/com/alibaba/dubbo/rpc/cluster/support/ClusterUtils.java b/dubbo-cluster/src/main/java/com/alibaba/dubbo/rpc/cluster/support/ClusterUtils.java
@@ -18,6 +18,7 @@
 
 import com.alibaba.dubbo.common.Constants;
 import com.alibaba.dubbo.common.URL;
+import com.alibaba.dubbo.common.utils.StringUtils;
 
 import java.util.HashMap;
 import java.util.Map;
@@ -88,18 +89,24 @@ public static URL mergeUrl(URL remoteUrl, Map<String, String> localMap) {
             if (remoteTimestamp != null && remoteTimestamp.length() > 0) {
                 map.put(Constants.REMOTE_TIMESTAMP_KEY, remoteMap.get(Constants.TIMESTAMP_KEY));
             }
+            String tag = remoteMap.get(Constants.TAG_KEY);
+            if (StringUtils.isNotEmpty(tag)) {
+                map.put(Constants.TAG_KEY, tag);
+            } else {
+                map.remove(Constants.TAG_KEY);
+            }
             // Combine filters and listeners on Provider and Consumer
             String remoteFilter = remoteMap.get(Constants.REFERENCE_FILTER_KEY);
             String localFilter = localMap.get(Constants.REFERENCE_FILTER_KEY);
             if (remoteFilter != null && remoteFilter.length() > 0
                     && localFilter != null && localFilter.length() > 0) {
-                localMap.put(Constants.REFERENCE_FILTER_KEY, remoteFilter + "," + localFilter);
+                map.put(Constants.REFERENCE_FILTER_KEY, remoteFilter + "," + localFilter);
             }
             String remoteListener = remoteMap.get(Constants.INVOKER_LISTENER_KEY);
             String localListener = localMap.get(Constants.INVOKER_LISTENER_KEY);
             if (remoteListener != null && remoteListener.length() > 0
                     && localListener != null && localListener.length() > 0) {
-                localMap.put(Constants.INVOKER_LISTENER_KEY, remoteListener + "," + localListener);
+                map.put(Constants.INVOKER_LISTENER_KEY, remoteListener + "," + localListener);
             }
         }
 

diff --git a/dubbo-rpc/dubbo-rpc-dubbo/src/main/java/com/alibaba/dubbo/rpc/protocol/dubbo/DubboCodec.java b/dubbo-rpc/dubbo-rpc-dubbo/src/main/java/com/alibaba/dubbo/rpc/protocol/dubbo/DubboCodec.java
@@ -22,7 +22,6 @@
 import com.alibaba.dubbo.common.io.UnsafeByteArrayInputStream;
 import com.alibaba.dubbo.common.logger.Logger;
 import com.alibaba.dubbo.common.logger.LoggerFactory;
-import com.alibaba.dubbo.common.serialize.ObjectInput;
 import com.alibaba.dubbo.common.serialize.ObjectOutput;
 import com.alibaba.dubbo.common.utils.ReflectUtils;
 import com.alibaba.dubbo.common.utils.StringUtils;
@@ -73,13 +72,12 @@ protected Object decodeBody(Channel channel, InputStream is, byte[] header) thro
             byte status = header[3];
             res.setStatus(status);
             try {
-                ObjectInput in = CodecSupport.deserialize(channel.getUrl(), is, proto);
                 if (status == Response.OK) {
                     Object data;
                     if (res.isHeartbeat()) {
-                        data = decodeHeartbeatData(channel, in);
+                        data = decodeHeartbeatData(channel,  CodecSupport.deserialize(channel.getUrl(), is, proto));
                     } else if (res.isEvent()) {
-                        data = decodeEventData(channel, in);
+                        data = decodeEventData(channel,  CodecSupport.deserialize(channel.getUrl(), is, proto));
                     } else {
                         DecodeableRpcResult result;
                         if (channel.getUrl().getParameter(
@@ -97,7 +95,7 @@ protected Object decodeBody(Channel channel, InputStream is, byte[] header) thro
                     }
                     res.setResult(data);
                 } else {
-                    res.setErrorMessage(in.readUTF());
+                    res.setErrorMessage(CodecSupport.deserialize(channel.getUrl(), is, proto).readUTF());
                 }
             } catch (Throwable t) {
                 if (log.isWarnEnabled()) {
@@ -117,11 +115,10 @@ protected Object decodeBody(Channel channel, InputStream is, byte[] header) thro
             }
             try {
                 Object data;
-                ObjectInput in = CodecSupport.deserialize(channel.getUrl(), is, proto);
                 if (req.isHeartbeat()) {
-                    data = decodeHeartbeatData(channel, in);
+                    data = decodeHeartbeatData(channel, CodecSupport.deserialize(channel.getUrl(), is, proto));
                 } else if (req.isEvent()) {
-                    data = decodeEventData(channel, in);
+                    data = decodeEventData(channel, CodecSupport.deserialize(channel.getUrl(), is, proto));
                 } else {
                     DecodeableRpcInvocation inv;
                     if (channel.getUrl().getParameter(