You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@rocketmq.apache.org by di...@apache.org on 2019/10/18 01:58:24 UTC
[rocketmq-client-go] branch master updated: fix(client):fix
subscribe method is not based on the last subscription and remove useless
client config GroupName (#251)
This is an automated email from the ASF dual-hosted git repository.
dinglei pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/rocketmq-client-go.git
The following commit(s) were added to refs/heads/master by this push:
new f73c8d2 fix(client):fix subscribe method is not based on the last subscription and remove useless client config GroupName (#251)
f73c8d2 is described below
commit f73c8d2cd6654d6b791f902ae77d758da21a330f
Author: rongtong <ji...@mails.ucas.ac.cn>
AuthorDate: Fri Oct 18 09:58:18 2019 +0800
fix(client):fix subscribe method is not based on the last subscription and remove useless client config GroupName (#251)
---
core/api.go | 2 --
core/api_test.go | 17 ++++++-----------
core/push_consumer.go | 7 +++----
3 files changed, 9 insertions(+), 17 deletions(-)
diff --git a/core/api.go b/core/api.go
index 9bb69a5..b0273e0 100644
--- a/core/api.go
+++ b/core/api.go
@@ -29,7 +29,6 @@ type ClientConfig struct {
GroupID string
NameServer string
NameServerDomain string
- GroupName string
InstanceName string
Credentials *SessionCredentials
LogC *LogConfig
@@ -41,7 +40,6 @@ func (config *ClientConfig) String() string {
str = strJoin(str, "GroupId", config.GroupID)
str = strJoin(str, "NameServer", config.NameServer)
str = strJoin(str, "NameServerDomain", config.NameServerDomain)
- str = strJoin(str, "GroupName", config.GroupName)
str = strJoin(str, "InstanceName", config.InstanceName)
if config.LogC != nil {
diff --git a/core/api_test.go b/core/api_test.go
index 0c09dae..d7e9bbb 100644
--- a/core/api_test.go
+++ b/core/api_test.go
@@ -26,7 +26,6 @@ func TestProducerConfig_String(t *testing.T) {
pConfig.GroupID = "testGroup"
pConfig.NameServer = "localhost:9876"
pConfig.NameServerDomain = "domain1"
- pConfig.GroupName = "producerGroupName"
pConfig.InstanceName = "testProducer"
pConfig.LogC = &LogConfig{
Path: "/rocketmq/log",
@@ -38,10 +37,9 @@ func TestProducerConfig_String(t *testing.T) {
pConfig.MaxMessageSize = 1024
pConfig.ProducerModel = CommonProducer
- expect := "ProducerConfig=[GroupId: testGroup, NameServer: localhost:9876, NameServerDomain: domain1, " +
- "GroupName: producerGroupName, InstanceName: testProducer, " +
- "LogConfig: {Path:/rocketmq/log FileNum:16 FileSize:1048576 Level:Debug}, S" +
- "endMsgTimeout: 30, CompressLevel: 4, MaxMessageSize: 1024, ProducerModel: CommonProducer, ]"
+ expect := "ProducerConfig=[GroupId: testGroup, NameServer: localhost:9876, NameServerDomain: domain1, InstanceName: testProducer, " +
+ "LogConfig: {Path:/rocketmq/log FileNum:16 FileSize:1048576 Level:Debug}, " +
+ "SendMsgTimeout: 30, CompressLevel: 4, MaxMessageSize: 1024, ProducerModel: CommonProducer, ]"
assert.Equal(t, expect, pConfig.String())
}
@@ -49,7 +47,6 @@ func TestPushConsumerConfig_String(t *testing.T) {
pcConfig := PushConsumerConfig{}
pcConfig.GroupID = "testGroup"
pcConfig.NameServer = "localhost:9876"
- pcConfig.GroupName = "consumerGroupName"
pcConfig.InstanceName = "testPushConsumer"
pcConfig.LogC = &LogConfig{
Path: "/rocketmq/log",
@@ -58,14 +55,13 @@ func TestPushConsumerConfig_String(t *testing.T) {
Level: LogLevelDebug}
pcConfig.ThreadCount = 4
pcConfig.MessageBatchMaxSize = 1024
- expect := "PushConsumerConfig=[GroupId: testGroup, NameServer: localhost:9876, GroupName: consumerGroupName, " +
+ expect := "PushConsumerConfig=[GroupId: testGroup, NameServer: localhost:9876, " +
"InstanceName: testPushConsumer, LogConfig: {Path:/rocketmq/log FileNum:16 FileSize:1048576 Level:Debug}, " +
"ThreadCount: 4, MessageBatchMaxSize: 1024, ]"
assert.Equal(t, expect, pcConfig.String())
pcConfig.NameServerDomain = "domain1"
- expect = "PushConsumerConfig=[GroupId: testGroup, NameServer: localhost:9876, NameServerDomain: domain1, " +
- "GroupName: consumerGroupName, InstanceName: testPushConsumer, " +
+ expect = "PushConsumerConfig=[GroupId: testGroup, NameServer: localhost:9876, NameServerDomain: domain1, InstanceName: testPushConsumer, " +
"LogConfig: {Path:/rocketmq/log FileNum:16 FileSize:1048576 Level:Debug}, ThreadCount: 4, MessageBatchMaxSize: 1024, ]"
assert.Equal(t, expect, pcConfig.String())
@@ -74,8 +70,7 @@ func TestPushConsumerConfig_String(t *testing.T) {
pcConfig.ConsumerModel = CoCurrently
pcConfig.MaxCacheMessageSize = 1024
pcConfig.MaxCacheMessageSizeInMB = 2048
- expect = "PushConsumerConfig=[GroupId: testGroup, NameServer: localhost:9876, NameServerDomain: domain1, " +
- "GroupName: consumerGroupName, InstanceName: testPushConsumer, " +
+ expect = "PushConsumerConfig=[GroupId: testGroup, NameServer: localhost:9876, NameServerDomain: domain1, InstanceName: testPushConsumer, " +
"LogConfig: {Path:/rocketmq/log FileNum:16 FileSize:1048576 Level:Debug}, ThreadCount: 4," +
" MessageBatchMaxSize: 32, MessageModel: Clustering, ConsumerModel: CoCurrently," +
" MaxCacheMessageSize: 1024, MaxCacheMessageSizeInMB: 2048, ]"
diff --git a/core/push_consumer.go b/core/push_consumer.go
index 32e47f3..bd7b68a 100644
--- a/core/push_consumer.go
+++ b/core/push_consumer.go
@@ -94,7 +94,7 @@ func newPushConsumer(config *PushConsumerConfig) (PushConsumer, error) {
C.free(unsafe.Pointer(cs))
if cconsumer == nil {
- return nil, errors.New("Create PushConsumer failed")
+ return nil, errors.New("create PushConsumer failed")
}
var err rmqError
@@ -242,9 +242,8 @@ func (c *defaultPushConsumer) Shutdown() error {
}
func (c *defaultPushConsumer) Subscribe(topic, expression string, consumeFunc func(msg *MessageExt) ConsumeStatus) error {
- _, exist := c.funcsMap.Load(topic)
- if exist {
- return nil
+ if consumeFunc == nil {
+ return errors.New("consumeFunc is nil")
}
err := rmqError(C.Subscribe(c.cconsumer, C.CString(topic), C.CString(expression)))
if err != NIL {