This is an automated email from the ASF dual-hosted git repository.
huzongtang pushed a commit to branch native
in repository https://gitbox.apache.org/repos/asf/rocketmq-client-go.git
The following commit(s) were added to refs/heads/native by this push:
new 36b0014 add broadcast consumer examples (#165)
36b0014 is described below
commit 36b001418eadf5e41fe9d4b40ff73aa1ce081771
Author: Liberxue <[email protected]>
AuthorDate: Mon Aug 26 10:54:45 2019 +0800
add broadcast consumer examples (#165)
* add broadcast consumer examples
* add consumer Shutdown
* Remove excess import
* fix go fmt ./... && [[ -z M examples/consumer/pull/main.go ]]
---
.DS_Store | Bin 0 -> 6148 bytes
examples/consumer/acl/main.go | 4 ++++
examples/consumer/{acl => broadcast}/main.go | 20 +++++++++-----------
examples/consumer/delay/main.go | 4 ++++
examples/consumer/interceptor/main.go | 4 ++++
examples/consumer/namespace/main.go | 4 ++++
examples/consumer/orderly/main.go | 4 ++++
examples/consumer/retry/concurrent/main.go | 4 ++++
examples/consumer/simple/main.go | 4 ++++
examples/consumer/strategy/main.go | 4 ++++
examples/consumer/trace/main.go | 5 +++++
11 files changed, 46 insertions(+), 11 deletions(-)
diff --git a/.DS_Store b/.DS_Store
new file mode 100644
index 0000000..7f3abed
Binary files /dev/null and b/.DS_Store differ
diff --git a/examples/consumer/acl/main.go b/examples/consumer/acl/main.go
index 8f944e7..940af52 100644
--- a/examples/consumer/acl/main.go
+++ b/examples/consumer/acl/main.go
@@ -56,5 +56,9 @@ func main() {
fmt.Println(err.Error())
os.Exit(-1)
}
+ err = c.Shutdown()
+ if err != nil {
+ fmt.Printf("Shutdown Consumer error: %s", err.Error())
+ }
time.Sleep(time.Hour)
}
diff --git a/examples/consumer/acl/main.go b/examples/consumer/broadcast/main.go
similarity index 82%
copy from examples/consumer/acl/main.go
copy to examples/consumer/broadcast/main.go
index 8f944e7..8f6fe1c 100644
--- a/examples/consumer/acl/main.go
+++ b/examples/consumer/broadcast/main.go
@@ -29,20 +29,13 @@ import (
)
func main() {
- c, err := rocketmq.NewPushConsumer(
+ c, _ := rocketmq.NewPushConsumer(
consumer.WithGroupName("testGroup"),
consumer.WithNameServer([]string{"127.0.0.1:9876"}),
- consumer.WithCredentials(primitive.Credentials{
- AccessKey: "RocketMQ",
- SecretKey: "12345678",
- }),
+ consumer.WithConsumeFromWhere(consumer.ConsumeFromFirstOffset),
+ consumer.WithConsumerModel(consumer.BroadCasting),
)
- if err != nil {
- fmt.Println("init consumer error: " + err.Error())
- os.Exit(0)
- }
-
- err = c.Subscribe("test", consumer.MessageSelector{}, func(ctx
context.Context,
+ err := c.Subscribe("min", consumer.MessageSelector{}, func(ctx
context.Context,
msgs ...*primitive.MessageExt) (consumer.ConsumeResult, error) {
fmt.Printf("subscribe callback: %v \n", msgs)
return consumer.ConsumeSuccess, nil
@@ -56,5 +49,10 @@ func main() {
fmt.Println(err.Error())
os.Exit(-1)
}
+ err = c.Shutdown()
+ if err != nil {
+ fmt.Printf("Shutdown Consumer error: %s", err.Error())
+ os.Exit(-1)
+ }
time.Sleep(time.Hour)
}
diff --git a/examples/consumer/delay/main.go b/examples/consumer/delay/main.go
index 2a639f9..6571e0f 100644
--- a/examples/consumer/delay/main.go
+++ b/examples/consumer/delay/main.go
@@ -52,5 +52,9 @@ func main() {
fmt.Println(err.Error())
os.Exit(-1)
}
+ err = c.Shutdown()
+ if err != nil {
+ fmt.Printf("Shutdown Consumer error: %s", err.Error())
+ }
time.Sleep(time.Hour)
}
diff --git a/examples/consumer/interceptor/main.go
b/examples/consumer/interceptor/main.go
index e4d14c8..e09e500 100644
--- a/examples/consumer/interceptor/main.go
+++ b/examples/consumer/interceptor/main.go
@@ -50,6 +50,10 @@ func main() {
os.Exit(-1)
}
time.Sleep(time.Hour)
+ err = c.Shutdown()
+ if err != nil {
+ fmt.Printf("Shutdown Consumer error: %s", err.Error())
+ }
}
func UserFistInterceptor() primitive.Interceptor {
diff --git a/examples/consumer/namespace/main.go
b/examples/consumer/namespace/main.go
index a0e8708..c2f356c 100644
--- a/examples/consumer/namespace/main.go
+++ b/examples/consumer/namespace/main.go
@@ -57,5 +57,9 @@ func main() {
fmt.Println(err.Error())
os.Exit(-1)
}
+ err = c.Shutdown()
+ if err != nil {
+ fmt.Printf("Shutdown Consumer error: %s", err.Error())
+ }
time.Sleep(time.Hour)
}
diff --git a/examples/consumer/orderly/main.go
b/examples/consumer/orderly/main.go
index f51bffd..a646cc0 100644
--- a/examples/consumer/orderly/main.go
+++ b/examples/consumer/orderly/main.go
@@ -52,5 +52,9 @@ func main() {
fmt.Println(err.Error())
os.Exit(-1)
}
+ err = c.Shutdown()
+ if err != nil {
+ fmt.Printf("Shutdown Consumer error: %s", err.Error())
+ }
time.Sleep(time.Hour)
}
diff --git a/examples/consumer/retry/concurrent/main.go
b/examples/consumer/retry/concurrent/main.go
index 8a21f83..52e5cf9 100644
--- a/examples/consumer/retry/concurrent/main.go
+++ b/examples/consumer/retry/concurrent/main.go
@@ -69,5 +69,9 @@ func main() {
fmt.Println(err.Error())
os.Exit(-1)
}
+ err = c.Shutdown()
+ if err != nil {
+ fmt.Printf("shundown Consumer error: %s", err.Error())
+ }
time.Sleep(time.Hour)
}
diff --git a/examples/consumer/simple/main.go b/examples/consumer/simple/main.go
index 2ddc465..61593b4 100644
--- a/examples/consumer/simple/main.go
+++ b/examples/consumer/simple/main.go
@@ -47,5 +47,9 @@ func main() {
fmt.Println(err.Error())
os.Exit(-1)
}
+ err = c.Shutdown()
+ if err != nil {
+ fmt.Printf("shundown Consumer error: %s", err.Error())
+ }
time.Sleep(time.Hour)
}
diff --git a/examples/consumer/strategy/main.go
b/examples/consumer/strategy/main.go
index 4483c0d..1858164 100644
--- a/examples/consumer/strategy/main.go
+++ b/examples/consumer/strategy/main.go
@@ -48,5 +48,9 @@ func main() {
fmt.Println(err.Error())
os.Exit(-1)
}
+ err = c.Shutdown()
+ if err != nil {
+ fmt.Printf("shundown Consumer error: %s", err.Error())
+ }
time.Sleep(time.Hour)
}
diff --git a/examples/consumer/trace/main.go b/examples/consumer/trace/main.go
index 23b09ab..87eb74f 100644
--- a/examples/consumer/trace/main.go
+++ b/examples/consumer/trace/main.go
@@ -53,6 +53,11 @@ func main() {
if err != nil {
fmt.Println(err.Error())
os.Exit(-1)
+
+ }
+ err = c.Shutdown()
+ if err != nil {
+ fmt.Printf("shundown Consumer error: %s", err.Error())
}
time.Sleep(time.Hour)
}
