messaging

Documentation

Index

• Constants
• Variables
• func NewContext(ctx context.Context, message *MessageMetadata) context.Context
• type DefaultMailboxProvider
  • func (p *DefaultMailboxProvider[TMessage]) GetMailbox(aggregateID string, aggregateTypeName string, ...) Mailbox[TMessage]
  • func (p *DefaultMailboxProvider[TMessage]) Stats() MailBoxStatistic
• type ExtensionKey
• type Extensions
  • func (extensions Extensions) Clone() Extensions
  • func (extensions Extensions) CustomExtensions() Extensions
  • func (extensions Extensions) Get(key ExtensionKey) (val string, ok bool)
  • func (extensions Extensions) Merge(toMerge Extensions) Extensions
  • func (extensions Extensions) Remove(keys ...ExtensionKey) Extensions
  • func (extensions Extensions) Set(key ExtensionKey, val string) Extensions
• type Mail
• type MailBoxStatistic
• type Mailbox
• type MailboxProvider
• type MailsWithResult
• type Message
• type MessageHandleFunc
• type MessageHandleResult
• type MessageHandleResultNotifier
• type MessageHandleResultWatchItem
• type MessageHandleResultWatcher
• type MessageHandler
• type MessageHandlerProxy
• type MessageMetadata
  • func FromContext(ctx context.Context) *MessageMetadata

Constants

const (
	SysExtensionKeyPrefix             string       = "squat:"
	ExtensionKeyNoticeServiceEndpoint ExtensionKey = ExtensionKey(SysExtensionKeyPrefix) + "NoticeServiceEndpoint"
	ExtensionKeyFromMessageID         ExtensionKey = ExtensionKey(SysExtensionKeyPrefix) + "FromMessageID"
	ExtensionKeyFromMessageType       ExtensionKey = ExtensionKey(SysExtensionKeyPrefix) + "FromMessageType"
	ExtensionKeyAggregateChanged      ExtensionKey = ExtensionKey(SysExtensionKeyPrefix) + "AggregateChanged"
)

const (
	ErrCodeMissingMessageHandler string = errors.SysErrCodePrefix + "MissingMessageHandler"
)

Variables

var (
	ErrMissingMessageHandler error = errors.NewWithCode(ErrCodeMissingMessageHandler, "missing message handler")
)

Functions

func NewContext

func NewContext(ctx context.Context, message *MessageMetadata) context.Context

Types

type DefaultMailboxProvider

type DefaultMailboxProvider[TMessage any] struct {
	MailboxCapacity    int
	GetMailboxName     func(aggregateID string, aggregateTypeName string) string
	AutoReleaseTimeout time.Duration
	// contains filtered or unexported fields
}

func (*DefaultMailboxProvider[TMessage]) GetMailbox

func (p *DefaultMailboxProvider[TMessage]) GetMailbox(aggregateID string, aggregateTypeName string, handlers map[string][]MessageHandler[TMessage]) Mailbox[TMessage]

func (*DefaultMailboxProvider[TMessage]) Stats

func (p *DefaultMailboxProvider[TMessage]) Stats() MailBoxStatistic

type ExtensionKey

type ExtensionKey string

type Extensions

type Extensions map[string]string

func (Extensions) Clone

func (extensions Extensions) Clone() Extensions

func (Extensions) CustomExtensions

func (extensions Extensions) CustomExtensions() Extensions

func (Extensions) Get

func (extensions Extensions) Get(key ExtensionKey) (val string, ok bool)

func (Extensions) Merge

func (extensions Extensions) Merge(toMerge Extensions) Extensions

func (Extensions) Remove

func (extensions Extensions) Remove(keys ...ExtensionKey) Extensions

func (Extensions) Set

func (extensions Extensions) Set(key ExtensionKey, val string) Extensions

type Mail

type Mail[TMessage any] interface {
	serialization.Serializable
	Metadata() MessageMetadata
	Unwrap() TMessage
}

type MailBoxStatistic

type MailBoxStatistic struct {
	CreatedCount int64
	RemovedCount int64
}

type Mailbox

type Mailbox[TMessageBody any] interface {
	Name() string
	SendMail(data MailsWithResult[TMessageBody]) error
}

type MailboxProvider

type MailboxProvider[TMessageBody any] interface {
	GetMailbox(aggregateID string, aggregateTypeName string, handlers map[string][]MessageHandler[TMessageBody]) Mailbox[TMessageBody]
	Stats() MailBoxStatistic
}

type MailsWithResult

type MailsWithResult[TMessage any] struct {
	Category string
	Mails    []Mail[TMessage]
	ResultCh chan MessageHandleResult
}

type Message

type Message struct {
	MessageMetadata
	Body string `json:"body"`
}

type MessageHandleFunc

type MessageHandleFunc[TMessage any] func(ctx context.Context, data TMessage) error

type MessageHandleResult

type MessageHandleResult struct {
	Err        error
	Extensions Extensions
}

type MessageHandleResultNotifier

type MessageHandleResultNotifier interface {
	Notify(endpoint string, messageID string, resultProvider string, result MessageHandleResult)
}

type MessageHandleResultWatchItem

type MessageHandleResultWatchItem interface {
	Result() <-chan MessageHandleResult
	Unwatch()
}

type MessageHandleResultWatcher

type MessageHandleResultWatcher interface {
	Watch(messageID string, resultProvider string) MessageHandleResultWatchItem
}

type MessageHandler

type MessageHandler[TMessage any] struct {
	Handle   MessageHandleFunc[TMessage]
	FuncName string
}

type MessageHandlerProxy

type MessageHandlerProxy[TMessage any] interface {
	Name() string
	Wrap(handleFuncName string, previousHandle MessageHandleFunc[TMessage]) MessageHandleFunc[TMessage]
}

type MessageMetadata

type MessageMetadata struct {
	MessageID     string     `json:"message_id"`
	MessageType   string     `json:"message_type"`
	AggregateID   string     `json:"aggregate_id"`
	AggregateType string     `json:"aggregate_type"`
	Category      string     `json:"category"`
	Extensions    Extensions `json:"extensions"`
}

func FromContext

func FromContext(ctx context.Context) *MessageMetadata