messaging

Documentation

Index

• type BaseMessage
  • func (m *BaseMessage) Addressee() string
  • func (m *BaseMessage) OpCode() int
  • func (m *BaseMessage) Payload() any
  • func (m *BaseMessage) SessionId() string
  • func (m *BaseMessage) Topic() string
• type IMessage
• type IMessageBus
  • func NewInMemoryMessageBus() (mq IMessageBus, err error)
• type IMessageProducer
• type InMemoryMessageBus
  • func (m *InMemoryMessageBus) Close() error
  • func (m *InMemoryMessageBus) CreateProducer(topic string) (IMessageProducer, error)
  • func (m *InMemoryMessageBus) Ping(retries uint, intervalInSeconds uint) error
  • func (m *InMemoryMessageBus) Pop(mf MessageFactory, timeout time.Duration, queue ...string) (IMessage, error)
  • func (m *InMemoryMessageBus) Publish(messages ...IMessage) error
  • func (m *InMemoryMessageBus) Push(messages ...IMessage) error
  • func (m *InMemoryMessageBus) Subscribe(mf MessageFactory, callback SubscriptionCallback, subscriber string, ...) (subscriptionId string, error error)
  • func (m *InMemoryMessageBus) Unsubscribe(subscriptionId string) (success bool)
• type MessageFactory
• type SubscriptionCallback

Types

type BaseMessage

type BaseMessage struct {
	MsgTopic     string `json:"topic"`     // Message topic (channel)
	MsgOpCode    int    `json:"opCode"`    // Message op code
	MsgAddressee string `json:"addressee"` // Message final addressee
	MsgSessionId string `json:"sessionId"` // Session id shared across all messages related to the same session
	MsgPayload   any    `json:"payload"`   // Message payload
}

BaseMessage basie implementation of IMessage interface

func (*BaseMessage) Addressee

func (m *BaseMessage) Addressee() string

func (*BaseMessage) OpCode

func (m *BaseMessage) OpCode() int

func (*BaseMessage) Payload

func (m *BaseMessage) Payload() any

func (*BaseMessage) SessionId

func (m *BaseMessage) SessionId() string

func (*BaseMessage) Topic

func (m *BaseMessage) Topic() string

type IMessage

type IMessage interface {
	// Topic name (also known as channel or queue)
	Topic() string

	// OpCode message operational code
	OpCode() int

	// Addressee message final addressee (recipient) - optional field
	Addressee() string

	// SessionId identifies a message exchange session which is shared across all messages related to the same session
	SessionId() string

	// Payload is the message body
	Payload() any
}

IMessage General message interface

type IMessageBus

type IMessageBus interface {

	// Closer includes method Close()
	io.Closer

	// Ping Test connectivity for retries number of time with time interval (in seconds) between retries
	Ping(retries uint, intervalInSeconds uint) error

	// Publish messages to a channel (topic)
	Publish(messages ...IMessage) error

	// Subscribe on topics and return subscriberId
	Subscribe(mf MessageFactory, callback SubscriptionCallback, subscriber string, topics ...string) (string, error)

	// Unsubscribe with the given subscriber id
	Unsubscribe(subscriptionId string) bool

	// Push Append one or multiple messages to a queue
	Push(messages ...IMessage) error

	// Pop Remove and get the last message in a queue or block until timeout expires
	Pop(mf MessageFactory, timeout time.Duration, queue ...string) (IMessage, error)

	// CreateProducer creates message producer for specific topic
	CreateProducer(topic string) (IMessageProducer, error)
}

IMessageBus Message bus interface

func NewInMemoryMessageBus

func NewInMemoryMessageBus() (mq IMessageBus, err error)

NewInMemoryMessageBus Factory method

type IMessageProducer

type IMessageProducer interface {

	// Closer includes method Close()
	io.Closer

	// Publish messages to a producer channel (topic)
	Publish(messages ...IMessage) error
}

IMessageProducer Message bus producer interface

type InMemoryMessageBus

type InMemoryMessageBus struct {
	// contains filtered or unexported fields
}

InMemoryMessageBus represents in memory implementation of IMessageBus interface topics is a map ot topic -> array of channels (channel per subscriber)

func (*InMemoryMessageBus) Close

func (m *InMemoryMessageBus) Close() error

Close client and free resources

func (*InMemoryMessageBus) CreateProducer

func (m *InMemoryMessageBus) CreateProducer(topic string) (IMessageProducer, error)

CreateProducer creates message producer for specific topic

func (*InMemoryMessageBus) Ping

func (m *InMemoryMessageBus) Ping(retries uint, intervalInSeconds uint) error

Ping Test connectivity for retries number of time with time interval (in seconds) between retries

func (*InMemoryMessageBus) Pop

func (m *InMemoryMessageBus) Pop(mf MessageFactory, timeout time.Duration, queue ...string) (IMessage, error)

Pop Remove and get the last message in a queue or block until timeout expires

func (*InMemoryMessageBus) Publish

func (m *InMemoryMessageBus) Publish(messages ...IMessage) error

Publish messages to a channel (topic)

func (*InMemoryMessageBus) Push

func (m *InMemoryMessageBus) Push(messages ...IMessage) error

Push Append one or multiple messages to a queue

func (*InMemoryMessageBus) Subscribe

func (m *InMemoryMessageBus) Subscribe(mf MessageFactory, callback SubscriptionCallback, subscriber string, topics ...string) (subscriptionId string, error error)

Subscribe on topics

func (*InMemoryMessageBus) Unsubscribe

func (m *InMemoryMessageBus) Unsubscribe(subscriptionId string) (success bool)

Unsubscribe with the given subscriber id

type MessageFactory

type MessageFactory func() IMessage

MessageFactory is a factory method of any message

type SubscriptionCallback

type SubscriptionCallback func(msg IMessage) bool

SubscriptionCallback Message subscription callback function, return true for ack