messaging

Documentation

Index

• Constants
• 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
  • func (m *BaseMessage) Version() string
• type EntityMessage
  • func (m *EntityMessage) Payload() any
• type IMessage
  • func NewEntityMessage() IMessage
• type IMessageBus
  • func NewInMemoryMessageBus() (mq IMessageBus, err error)
• type IMessageConsumer
• type IMessageProducer
• type InMemoryMessageBus
  • func (m *InMemoryMessageBus) Close() error
  • func (m *InMemoryMessageBus) CreateConsumer(subscription string, mf MessageFactory, topics ...string) (IMessageConsumer, 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(subscription string, mf MessageFactory, callback SubscriptionCallback, ...) (subscriptionId string, error error)
  • func (m *InMemoryMessageBus) Unsubscribe(subscriptionId string) (success bool)
• type InMemoryMessageConsumer
  • func (m *InMemoryMessageConsumer) Close() error
  • func (m *InMemoryMessageConsumer) Read(timeout time.Duration) (IMessage, error)
• type MessageFactory
• type SubscriptionCallback

Constants

const EntityMessageTopic = "ENTITY"

Types

type BaseMessage

type BaseMessage struct {
	MsgTopic     string `json:"topic"`     // Message topic (channel)
	MsgOpCode    int    `json:"opCode"`    // Message op code
	MsgVersion   string `json:"version"`   // 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
}

BaseMessage base 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

func (*BaseMessage) Version

func (m *BaseMessage) Version() string

type EntityMessage

type EntityMessage struct {
	BaseMessage
	MsgPayload any `json:"payload"` // Payload
}

EntityMessage generic implementation of IMessage interface

func (*EntityMessage) Payload

func (m *EntityMessage) Payload() any

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

	// Version identifies a message version
	Version() string

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

IMessage General message interface

func NewEntityMessage

func NewEntityMessage() IMessage

NewEntityMessage is a message factory

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(subscription string, mf MessageFactory, callback SubscriptionCallback, 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 a specific topic
	CreateProducer(topic string) (IMessageProducer, error)

	// CreateConsumer creates message consumer for a specific topic
	CreateConsumer(subscription string, mf MessageFactory, topics ...string) (IMessageConsumer, error)
}

IMessageBus Message bus interface

func NewInMemoryMessageBus

func NewInMemoryMessageBus() (mq IMessageBus, err error)

NewInMemoryMessageBus Factory method

type IMessageConsumer

type IMessageConsumer interface {
	// Closer includes method Close()
	io.Closer

	// Read message from topic, blocks until a new message arrive or until timeout expires
	// Use 0 instead of time.Duration for unlimited time
	// The standard way to use Read is by using infinite loop:
	//
	//	for {
	//		if msg, err := consumer.Read(time.Second * 5); err != nil {
	//			// Handle error
	//		} else {
	//			// Process message in a dedicated go routine
	//			go processTisMessage(msg)
	//		}
	//	}
	//
	Read(timeout time.Duration) (IMessage, error)
}

IMessageConsumer Message bus consumer (reader) interface

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) CreateConsumer

func (m *InMemoryMessageBus) CreateConsumer(subscription string, mf MessageFactory, topics ...string) (IMessageConsumer, error)

CreateConsumer creates message consumer for a specific topic

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(subscription string, mf MessageFactory, callback SubscriptionCallback, 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 InMemoryMessageConsumer

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

func (*InMemoryMessageConsumer) Close

func (m *InMemoryMessageConsumer) Close() error

Close client and free resources

func (*InMemoryMessageConsumer) Read

func (m *InMemoryMessageConsumer) Read(timeout time.Duration) (IMessage, error)

Read message from topic, blocks until a new message arrive or until timeout

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