summaryrefslogtreecommitdiff
path: root/pkg/pubsub
diff options
context:
space:
mode:
Diffstat (limited to 'pkg/pubsub')
-rw-r--r--pkg/pubsub/interface.go14
-rw-r--r--pkg/pubsub/message.go31
2 files changed, 10 insertions, 35 deletions
diff --git a/pkg/pubsub/interface.go b/pkg/pubsub/interface.go
index 80dab0c3..caf8783f 100644
--- a/pkg/pubsub/interface.go
+++ b/pkg/pubsub/interface.go
@@ -11,6 +11,7 @@ type PubSub interface {
type Subscriber interface {
// Subscribe broker to one or multiple topics.
Subscribe(topics ...string) error
+
// Unsubscribe from one or multiply topics
Unsubscribe(topics ...string) error
}
@@ -18,21 +19,14 @@ type Subscriber interface {
// Publisher publish one or more messages
type Publisher interface {
// Publish one or multiple Channel.
- Publish(messages []Message) error
+ Publish(messages []*Message) error
// PublishAsync publish message and return immediately
// If error occurred it will be printed into the logger
- PublishAsync(messages []Message)
+ PublishAsync(messages []*Message)
}
// Reader interface should return next message
type Reader interface {
- Next() (Message, error)
-}
-
-type Message interface {
- Command() string
- Payload() []byte
- Topics() []string
- Broker() string
+ Next() (*Message, error)
}
diff --git a/pkg/pubsub/message.go b/pkg/pubsub/message.go
index 17e6780f..c1a7246a 100644
--- a/pkg/pubsub/message.go
+++ b/pkg/pubsub/message.go
@@ -4,40 +4,21 @@ import (
json "github.com/json-iterator/go"
)
-type Msg struct {
+type Message struct {
// Topic message been pushed into.
- Topics_ []string `json:"topic"`
+ Topics []string `json:"topic"`
// Command (join, leave, headers)
- Command_ string `json:"command"`
+ Command string `json:"command"`
// Broker (redis, memory)
- Broker_ string `json:"broker"`
+ Broker string `json:"broker"`
// Payload to be broadcasted
- Payload_ []byte `json:"payload"`
+ Payload []byte `json:"payload"`
}
// MarshalBinary needed to marshal message for the redis
-func (m *Msg) MarshalBinary() ([]byte, error) {
+func (m *Message) MarshalBinary() ([]byte, error) {
return json.Marshal(m)
}
-
-// Payload in raw bytes
-func (m *Msg) Payload() []byte {
- return m.Payload_
-}
-
-// Command for the connection
-func (m *Msg) Command() string {
- return m.Command_
-}
-
-// Topics to subscribe
-func (m *Msg) Topics() []string {
- return m.Topics_
-}
-
-func (m *Msg) Broker() string {
- return m.Broker_
-}