add topic filter to handle incoming messages for specific topic

unit-adm · unit-adm · commit c6b9175a582a · 2021-08-20T17:50:43.000+05:30
diff --git a/README.md b/README.md
@@ -16,7 +16,7 @@ Make use of the client by importing it in your Go client source code. For exampl
 
 import "github.com/unit-io/unitdb-go"
 
-Samples are available in the cmd directory for reference. To build unitdb server from latest source code use "replace" in go.mod to point to your local module.
+Samples are available in the examples directory for reference. To build unitdb server from latest source code use "replace" in go.mod to point to your local module.
 
 ```golang
 go mod edit -replace github.com/unit-io/unitdb=$GOPATH/src/github.com/unit-io/unitdb
diff --git a/client.go b/client.go
@@ -327,7 +327,7 @@ func (c *client) TopicFilter(subscriptionTopic string) (*TopicFilter, error) {
 		validateTopicParts); err != nil {
 		return nil, err
 	}
-	t := &TopicFilter{subscriptionTopic: topic, updates: make(chan []PubMessage)}
+	t := &TopicFilter{subscriptionTopic: topic, updates: make(chan []*PubMessage)}
 	c.notifier.addFilter(t.filter)
 
 	return t, nil
diff --git a/go.mod b/go.mod
@@ -4,8 +4,8 @@ go 1.16
 
 require (
 	github.com/golang/protobuf v1.5.2
-	github.com/unit-io/unitdb v0.1.1
+	github.com/unit-io/unitdb v0.2.0
 	google.golang.org/grpc v1.39.0
 )
 
-// replace github.com/unit-io/unitdb => /src/github.com/unit-io/unitdb
+// replace github.com/unit-io/unitdb => github.com/unit-io/unitdb
diff --git a/go.sum b/go.sum
@@ -14,7 +14,6 @@ github.com/envoyproxy/go-control-plane v0.9.0/go.mod h1:YTl/9mNaCwkRvm6d1a2C3ymF
 github.com/envoyproxy/go-control-plane v0.9.1-0.20191026205805-5f8ba28d4473/go.mod h1:YTl/9mNaCwkRvm6d1a2C3ymFceY/DCBVvsKhRF0iEA4=
 github.com/envoyproxy/go-control-plane v0.9.4/go.mod h1:6rpuAdCZL397s3pYoYcLgu1mIlRU8Am5FuJP05cCM98=
 github.com/envoyproxy/go-control-plane v0.9.9-0.20201210154907-fd9021fe5dad/go.mod h1:cXg6YxExXjJnVBQHBLXeUAgxn2UodCpnH306RInaBQk=
-github.com/envoyproxy/go-control-plane v0.9.9-0.20210217033140-668b12f5399d/go.mod h1:cXg6YxExXjJnVBQHBLXeUAgxn2UodCpnH306RInaBQk=
 github.com/envoyproxy/go-control-plane v0.9.9-0.20210512163311-63b5d3c536b0/go.mod h1:hliV/p42l8fGbc6Y9bQ70uLwIvmJyVE5k4iMKlh8wCQ=
 github.com/envoyproxy/protoc-gen-validate v0.1.0/go.mod h1:iSmxcyjqTsJpI2R4NaDN7+kN2VEUnK/pcBlmesArF7c=
 github.com/ghodss/yaml v1.0.0/go.mod h1:4dBDuWmgqj2HViK6kFavaiC9ZROes6MMH2rRYeMEF04=
@@ -55,9 +54,8 @@ github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+
 github.com/stretchr/testify v1.5.1/go.mod h1:5W2xD1RspED5o8YsWQXVCued0rvSQ+mT+I5cxcmMvtA=
 github.com/unit-io/bpool v0.0.0-20200906005724-1643bbf59264 h1:31MK/k8NNVPI3UEUCJ8+9aEzlFxJRhawKk2gqxSbJeA=
 github.com/unit-io/bpool v0.0.0-20200906005724-1643bbf59264/go.mod h1:jLqAtkF257MDiAc5K8svPVUGjfig2qdIhnWs3OCDwKg=
-github.com/unit-io/unitdb v0.1.1 h1:KzTFWpsDfyo2qV1AYpoHmoI2FGMJY74KncvZM8CAgHU=
-github.com/unit-io/unitdb v0.1.1/go.mod h1:B8CPnLiFt1OcRxWIj+69kNxMRuSoS7o3muBzgqLZbkk=
-github.com/unit-io/unitdb-go v0.0.0-20210407101657-d9db9270f78d/go.mod h1:HU0IfRu0nXM2O9vMCRhDhvBUBc604AMmCDwWEG5C7zk=
+github.com/unit-io/unitdb v0.2.0 h1:EWjq1JfAtrLZU83ka+wpS+aam39GN47bc1oxlvKETRI=
+github.com/unit-io/unitdb v0.2.0/go.mod h1:E2yJHO6xjTwqRmOwKO0BcbKpTj6oHM5xldD3nL4NP5Q=
 github.com/yuin/goldmark v1.2.1/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9decYSb74=
 go.opentelemetry.io/proto/otlp v0.7.0/go.mod h1:PqfVotwruBrMGOCsRd/89rSnXhoiJIqeYNgFYFoEGnI=
 golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w=
@@ -126,7 +124,6 @@ google.golang.org/grpc v1.25.1/go.mod h1:c3i+UQWmh7LiEpx4sFZnkU36qjEYZ0imhYfXVyQ
 google.golang.org/grpc v1.27.0/go.mod h1:qbnxyOmOxrQa7FizSgH+ReBfzJrCY1pSN7KXBS8abTk=
 google.golang.org/grpc v1.33.1/go.mod h1:fr5YgcSWrqhRRxogOsw7RzIpsmvOZ6IcH4kBYTpR3n0=
 google.golang.org/grpc v1.36.0/go.mod h1:qjiiYl8FncCW8feJPdyg3v6XW24KsRHe+dy9BAGRRjU=
-google.golang.org/grpc v1.37.0/go.mod h1:NREThFqKR1f3iQ6oBuvc5LadQuXVGo9rkm5ZGrQdJfM=
 google.golang.org/grpc v1.39.0 h1:Klz8I9kdtkIN6EpHHUOMLCYhTn/2WAe5a0s1hcBkdTI=
 google.golang.org/grpc v1.39.0/go.mod h1:PImNr+rS9TWYb2O4/emRugxiyHZ5JyHW5F+RPnDzfrE=
 google.golang.org/protobuf v0.0.0-20200109180630-ec00e32a8dfd/go.mod h1:DFci5gLYBciE7Vtevhsrf46CRTquxDuWsQurQQe4oz8=
diff --git a/message.go b/message.go
@@ -13,7 +13,7 @@ import (
 type Message interface {
 	DeliveryMode() uint8
 	MessageID() uint16
-	Messages() []PubMessage
+	Messages() []*PubMessage
 	Ack()
 }
 
@@ -25,7 +25,7 @@ type PubMessage struct {
 type message struct {
 	deliveryMode uint8
 	messageID    uint16
-	messages     []PubMessage
+	messages     []*PubMessage
 	once         sync.Once
 	ack          func()
 }
@@ -38,7 +38,7 @@ func (m *message) MessageID() uint16 {
 	return m.messageID
 }
 
-func (m *message) Messages() []PubMessage {
+func (m *message) Messages() []*PubMessage {
 	return m.messages
 }
 
@@ -47,9 +47,9 @@ func (m *message) Ack() {
 }
 
 func messageFromPublish(p *utp.Publish, ack func()) *message {
-	var messages []PubMessage
+	var messages []*PubMessage
 	for _, pubMsg := range p.Messages {
-		msg := PubMessage{
+		msg := &PubMessage{
 			Topic:   pubMsg.Topic,
 			Payload: pubMsg.Payload,
 		}
diff --git a/notifier.go b/notifier.go
@@ -30,7 +30,7 @@ const (
 type filter func(notice *Notice) error
 
 type Notice struct {
-	messages []PubMessage
+	messages []*PubMessage
 }
 
 // func (n *Notice) Messages() []PubMessage {
@@ -85,7 +85,7 @@ func (n *notifier) hasObservers() bool {
 	return len(n.filters) > 0
 }
 
-func (n *notifier) notify(messages []PubMessage) {
+func (n *notifier) notify(messages []*PubMessage) {
 	if !n.hasObservers() {
 		return
 	}
diff --git a/topic.go b/topic.go
@@ -52,11 +52,11 @@ type (
 	}
 	TopicFilter struct {
 		subscriptionTopic *topic
-		updates           chan []PubMessage
+		updates           chan []*PubMessage
 	}
 )
 
-func (t *TopicFilter) Updates() <-chan []PubMessage {
+func (t *TopicFilter) Updates() <-chan []*PubMessage {
 	return t.updates
 }
 
@@ -226,8 +226,7 @@ func validateTopicParts(t *topic) error {
 }
 
 func (t *TopicFilter) filter(notice *Notice) error {
-	fmt.Println("filter: message count", len(notice.messages))
-	messages := make([]PubMessage, 0)
+	messages := make([]*PubMessage, 0)
 
 	for _, pubMsg := range notice.messages {
 		pubTopic := new(topic)
@@ -246,13 +245,11 @@ func (t *TopicFilter) filter(notice *Notice) error {
 		}
 
 		if t.subscriptionTopic.matches(pubTopic) {
-			fmt.Println("matches: subTopic, pubTopic ", t.subscriptionTopic.topic, pubTopic.topic)
 			messages = append(messages, notice.messages...)
 		}
 	}
 
 	if len(messages) > 0 {
-		fmt.Println("filtered: message count", len(messages))
 		t.updates <- messages
 	}
 

Original file line number	Diff line number	Diff line change
`@@ -327,7 +327,7 @@ func (c client) TopicFilter(subscriptionTopic string) (TopicFilter, error) {`
`327`	`327`	`validateTopicParts); err != nil {`
`328`	`328`	`return nil, err`
`329`	`329`	`}`
`330`		`- t := &TopicFilter{subscriptionTopic: topic, updates: make(chan []PubMessage)}`
	`330`	`+ t := &TopicFilter{subscriptionTopic: topic, updates: make(chan []*PubMessage)}`
`331`	`331`	`c.notifier.addFilter(t.filter)`
`332`	`332`
`333`	`333`	`return t, nil`
Original file line number	Diff line number	Diff line change
`@@ -4,8 +4,8 @@ go 1.16`
`4`	`4`
`5`	`5`	`require (`
`6`	`6`	`github.com/golang/protobuf v1.5.2`
`7`		`- github.com/unit-io/unitdb v0.1.1`
	`7`	`+ github.com/unit-io/unitdb v0.2.0`
`8`	`8`	`google.golang.org/grpc v1.39.0`
`9`	`9`	`)`
`10`	`10`
`11`		`-// replace github.com/unit-io/unitdb => /src/github.com/unit-io/unitdb`
	`11`	`+// replace github.com/unit-io/unitdb => github.com/unit-io/unitdb`
Original file line number	Diff line number	Diff line change
`@@ -13,7 +13,7 @@ import (`
`13`	`13`	`type Message interface {`
`14`	`14`	`DeliveryMode() uint8`
`15`	`15`	`MessageID() uint16`
`16`		`- Messages() []PubMessage`
	`16`	`+ Messages() []*PubMessage`
`17`	`17`	`Ack()`
`18`	`18`	`}`
`19`	`19`
`@@ -25,7 +25,7 @@ type PubMessage struct {`
`25`	`25`	`type message struct {`
`26`	`26`	`deliveryMode uint8`
`27`	`27`	`messageID uint16`
`28`		`- messages []PubMessage`
	`28`	`+ messages []*PubMessage`
`29`	`29`	`once sync.Once`
`30`	`30`	`ack func()`
`31`	`31`	`}`
`@@ -38,7 +38,7 @@ func (m *message) MessageID() uint16 {`
`38`	`38`	`return m.messageID`
`39`	`39`	`}`
`40`	`40`
`41`		`-func (m *message) Messages() []PubMessage {`
	`41`	`+func (m message) Messages() []PubMessage {`
`42`	`42`	`return m.messages`
`43`	`43`	`}`
`44`	`44`
`@@ -47,9 +47,9 @@ func (m *message) Ack() {`
`47`	`47`	`}`
`48`	`48`
`49`	`49`	`func messageFromPublish(p utp.Publish, ack func()) message {`
`50`		`- var messages []PubMessage`
	`50`	`+ var messages []*PubMessage`
`51`	`51`	`for _, pubMsg := range p.Messages {`
`52`		`- msg := PubMessage{`
	`52`	`+ msg := &PubMessage{`
`53`	`53`	`Topic: pubMsg.Topic,`
`54`	`54`	`Payload: pubMsg.Payload,`
`55`	`55`	`}`
Original file line number	Diff line number	Diff line change
`@@ -30,7 +30,7 @@ const (`
`30`	`30`	`type filter func(notice *Notice) error`
`31`	`31`
`32`	`32`	`type Notice struct {`
`33`		`- messages []PubMessage`
	`33`	`+ messages []*PubMessage`
`34`	`34`	`}`
`35`	`35`
`36`	`36`	`// func (n *Notice) Messages() []PubMessage {`
`@@ -85,7 +85,7 @@ func (n *notifier) hasObservers() bool {`
`85`	`85`	`return len(n.filters) > 0`
`86`	`86`	`}`
`87`	`87`
`88`		`-func (n *notifier) notify(messages []PubMessage) {`
	`88`	`+func (n notifier) notify(messages []PubMessage) {`
`89`	`89`	`if !n.hasObservers() {`
`90`	`90`	`return`
`91`	`91`	`}`
Original file line number	Diff line number	Diff line change
`@@ -52,11 +52,11 @@ type (`
`52`	`52`	`}`
`53`	`53`	`TopicFilter struct {`
`54`	`54`	`subscriptionTopic *topic`
`55`		`- updates chan []PubMessage`
	`55`	`+ updates chan []*PubMessage`
`56`	`56`	`}`
`57`	`57`	`)`
`58`	`58`
`59`		`-func (t *TopicFilter) Updates() <-chan []PubMessage {`
	`59`	`+func (t TopicFilter) Updates() <-chan []PubMessage {`
`60`	`60`	`return t.updates`
`61`	`61`	`}`
`62`	`62`
`@@ -226,8 +226,7 @@ func validateTopicParts(t *topic) error {`
`226`	`226`	`}`
`227`	`227`
`228`	`228`	`func (t TopicFilter) filter(notice Notice) error {`
`229`		`- fmt.Println("filter: message count", len(notice.messages))`
`230`		`- messages := make([]PubMessage, 0)`
	`229`	`+ messages := make([]*PubMessage, 0)`
`231`	`230`
`232`	`231`	`for _, pubMsg := range notice.messages {`
`233`	`232`	`pubTopic := new(topic)`
`@@ -246,13 +245,11 @@ func (t TopicFilter) filter(notice Notice) error {`
`246`	`245`	`}`
`247`	`246`
`248`	`247`	`if t.subscriptionTopic.matches(pubTopic) {`
`249`		`- fmt.Println("matches: subTopic, pubTopic ", t.subscriptionTopic.topic, pubTopic.topic)`
`250`	`248`	`messages = append(messages, notice.messages...)`
`251`	`249`	`}`
`252`	`250`	`}`
`253`	`251`
`254`	`252`	`if len(messages) > 0 {`
`255`		`- fmt.Println("filtered: message count", len(messages))`
`256`	`253`	`t.updates <- messages`
`257`	`254`	`}`
`258`	`255`