test/milvus
1
0
Fork 0
mirror of https://gitee.com/milvus-io/milvus.git synced 2024-11-30 02:48:45 +08:00
Code Issues Actions 11 Packages Projects Releases Wiki Activity

[skip ci] Add comments for pulsar_consumer.go (#12446)

Browse Source 
Signed-off-by: shaoyue.chen <shaoyue.chen@zilliz.com>
This commit is contained in:
shaoyue 2021-11-30 16:05:59 +08:00 committed by GitHub
parent 5bffbba296
commit 2b3762fa63
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
1 changed files with 5 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

5
internal/util/mqclient/pulsar_consumer.go
View File

@ -19,6 +19,7 @@ import (
"github.com/milvus-io/milvus/internal/log"
)
// PulsarConsumer consumes from pulsar
type PulsarConsumer struct {
c pulsar.Consumer
pulsar.Reader
@ -30,10 +31,12 @@ type PulsarConsumer struct {
skip bool
}
// Subscription get a subscription for the consumer
func (pc *PulsarConsumer) Subscription() string {
return pc.c.Subscription()
}
// Chan returns a message channel
func (pc *PulsarConsumer) Chan() <-chan Message {
if pc.msgChannel == nil {
pc.once.Do(func() {
@ -88,11 +91,13 @@ func (pc *PulsarConsumer) Seek(id MessageID, inclusive bool) error {
return err
}
// Ack the consumption of a single message
func (pc *PulsarConsumer) Ack(message Message) {
pm := message.(*pulsarMessage)
pc.c.Ack(pm.msg)
}
// Close the consumer and stop the broker to push more messages
func (pc *PulsarConsumer) Close() {
pc.c.Close()
close(pc.closeCh)