commit 66e15cfa11531c8f09045ab5a2cf4c7a55aa3a50
author ShannonDing <libya_003@163.com> Thu Feb 28 17:50:52 2019 +0800
committer ShannonDing <libya_003@163.com> Thu Feb 28 17:50:52 2019 +0800
tree 6db7525598b429fec653abb295a0a64d0f62fa64
parent 4dbf4160d952b3d2ee8b91c7c6fa26e662f5c496

Send subscription to snode when using real push

client/src/main/java/org/apache/rocketmq/client/impl/consumer/DefaultMQRealPushConsumerImpl.java

diff --git a/client/src/main/java/org/apache/rocketmq/client/impl/consumer/DefaultMQRealPushConsumerImpl.java b/client/src/main/java/org/apache/rocketmq/client/impl/consumer/DefaultMQRealPushConsumerImpl.java
index 755ed58..e684dd1 100644
--- a/client/src/main/java/org/apache/rocketmq/client/impl/consumer/DefaultMQRealPushConsumerImpl.java
+++ b/client/src/main/java/org/apache/rocketmq/client/impl/consumer/DefaultMQRealPushConsumerImpl.java
@@ -914,10 +914,14 @@
                 topic, subExpression);
             this.rebalanceImpl.getSubscriptionInner().put(topic, subscriptionData);
             if (this.mQClientFactory != null) {
-                this.mQClientFactory.sendHeartbeatToAllBrokerWithLock();
+                if (realPushModel) {
+                    this.mQClientFactory.sendHeartbeatToAllSnodeWithLock();
+                } else {
+                    this.mQClientFactory.sendHeartbeatToAllBrokerWithLock();
+                }
             }
         } catch (Exception e) {
-            throw new MQClientException("subscription exception", e);
+            throw new MQClientException("Subscription exception", e);
         }
     }
 
@@ -930,11 +934,15 @@
             subscriptionData.setFilterClassSource(filterClassSource);
             this.rebalanceImpl.getSubscriptionInner().put(topic, subscriptionData);
             if (this.mQClientFactory != null) {
-                this.mQClientFactory.sendHeartbeatToAllBrokerWithLock();
+                if (realPushModel) {
+                    this.mQClientFactory.sendHeartbeatToAllSnodeWithLock();
+                } else {
+                    this.mQClientFactory.sendHeartbeatToAllBrokerWithLock();
+                }
             }
 
         } catch (Exception e) {
-            throw new MQClientException("subscription exception", e);
+            throw new MQClientException("Subscription exception", e);
         }
     }
 
@@ -950,7 +958,11 @@
 
             this.rebalanceImpl.getSubscriptionInner().put(topic, subscriptionData);
             if (this.mQClientFactory != null) {
-                this.mQClientFactory.sendHeartbeatToAllBrokerWithLock();
+                if (realPushModel) {
+                    this.mQClientFactory.sendHeartbeatToAllSnodeWithLock();
+                } else {
+                    this.mQClientFactory.sendHeartbeatToAllBrokerWithLock();
+                }
             }
         } catch (Exception e) {
             throw new MQClientException("subscription exception", e);