queues

Oct 7, 2022

Changes in this version

+ func GetMessageAs(envelope *MessageEnvelope) (T, error)

+ type CallbackMessageReceiver struct

+ Callback func(message *MessageEnvelope, queue IMessageQueue) error

+ func NewCallbackMessageReceiver(callback func(message *MessageEnvelope, queue IMessageQueue) error) *CallbackMessageReceiver

+ func (c *CallbackMessageReceiver) ReceiveMessage(message *MessageEnvelope, queue IMessageQueue) (err error)

+ type IMessageQueue interface

+ Abandon func(ctx context.Context, message *MessageEnvelope) error

+ BeginListen func(ctx context.Context, correlationId string, receiver IMessageReceiver)

+ Capabilities func() *MessagingCapabilities

+ Complete func(ctx context.Context, message *MessageEnvelope) error

+ EndListen func(ctx context.Context, correlationId string)

+ Listen func(ctx context.Context, correlationId string, receiver IMessageReceiver) error

+ MoveToDeadLetter func(ctx context.Context, message *MessageEnvelope) error

+ Name func() string

+ Peek func(ctx context.Context, correlationId string) (result *MessageEnvelope, err error)

+ PeekBatch func(ctx context.Context, correlationId string, messageCount int64) (result []*MessageEnvelope, err error)

+ ReadMessageCount func() (count int64, err error)

+ Receive func(ctx context.Context, correlationId string, waitTimeout time.Duration) (result *MessageEnvelope, err error)

+ RenewLock func(ctx context.Context, message *MessageEnvelope, lockTimeout time.Duration) error

+ Send func(ctx context.Context, correlationId string, envelope *MessageEnvelope) error

+ SendAsObject func(ctx context.Context, correlationId string, messageType string, value any) error

+ type IMessageQueueOverrides interface

+ OpenWithParams func(ctx context.Context, correlationId string, ...) error

+ type IMessageReceiver interface

+ ReceiveMessage func(ctx context.Context, envelope *MessageEnvelope, queue IMessageQueue) (err error)

+ type LockedMessage struct

+ ExpirationTime time.Time

+ Message *MessageEnvelope

+ Timeout time.Duration

+ type MemoryMessageQueue struct

+ func NewMemoryMessageQueue(name string) *MemoryMessageQueue

+ func (c *MemoryMessageQueue) Abandon(ctx context.Context, message *MessageEnvelope) (err error)

+ func (c *MemoryMessageQueue) Clear(ctx context.Context, correlationId string) (err error)

+ func (c *MemoryMessageQueue) Close(ctx context.Context, correlationId string) (err error)

+ func (c *MemoryMessageQueue) Complete(ctx context.Context, message *MessageEnvelope) (err error)

+ func (c *MemoryMessageQueue) EndListen(ctx context.Context, correlationId string)

+ func (c *MemoryMessageQueue) IsOpen() bool

+ func (c *MemoryMessageQueue) Listen(ctx context.Context, correlationId string, receiver IMessageReceiver) error

+ func (c *MemoryMessageQueue) MoveToDeadLetter(ctx context.Context, message *MessageEnvelope) (err error)

+ func (c *MemoryMessageQueue) Open(ctx context.Context, correlationId string) (err error)

+ func (c *MemoryMessageQueue) Peek(ctx context.Context, correlationId string) (result *MessageEnvelope, err error)

+ func (c *MemoryMessageQueue) PeekBatch(ctx context.Context, correlationId string, messageCount int64) (result []*MessageEnvelope, err error)

+ func (c *MemoryMessageQueue) ReadMessageCount() (count int64, err error)

+ func (c *MemoryMessageQueue) Receive(ctx context.Context, correlationId string, waitTimeout time.Duration) (*MessageEnvelope, error)

+ func (c *MemoryMessageQueue) RenewLock(ctx context.Context, message *MessageEnvelope, lockTimeout time.Duration) (err error)

+ func (c *MemoryMessageQueue) Send(ctx context.Context, correlationId string, envelope *MessageEnvelope) (err error)

+ type MessageEnvelope struct

+ CorrelationId string

+ JsonMapConvertor cconv.IJSONEngine[map[string]any]

+ Message []byte

+ MessageId string

+ MessageType string

+ SentTime time.Time

+ func NewEmptyMessageEnvelope() *MessageEnvelope

+ func NewMessageEnvelope(correlationId string, messageType string, message []byte) *MessageEnvelope

+ func NewMessageEnvelopeFromObject(correlationId string, messageType string, message any) *MessageEnvelope

+ func (c *MessageEnvelope) GetMessageAsString() string

+ func (c *MessageEnvelope) GetReference() any

+ func (c *MessageEnvelope) SetMessageAsObject(value any)

+ func (c *MessageEnvelope) SetMessageAsString(value string)

+ func (c *MessageEnvelope) SetReference(value any)

+ func (c *MessageEnvelope) String() string

+ func (c *MessageEnvelope) UnmarshalJSON(data []byte) error

+ func (c MessageEnvelope) MarshalJSON() ([]byte, error)

+ type MessageQueue struct

+ ConnectionResolver *cconn.ConnectionResolver

+ Counters *ccount.CompositeCounters

+ CredentialResolver *cauth.CredentialResolver

+ Lock sync.Mutex

+ Logger *clog.CompositeLogger

+ Overrides IMessageQueueOverrides

+ func InheritMessageQueue(overrides IMessageQueueOverrides, name string, ...) *MessageQueue

+ func (c *MessageQueue) BeginListen(ctx context.Context, correlationId string, receiver IMessageReceiver)

+ func (c *MessageQueue) Capabilities() *MessagingCapabilities

+ func (c *MessageQueue) CheckOpen(correlationId string) error

+ func (c *MessageQueue) Configure(ctx context.Context, config *cconf.ConfigParams)

+ func (c *MessageQueue) Name() string

+ func (c *MessageQueue) Open(ctx context.Context, correlationId string) error

+ func (c *MessageQueue) OpenWithParams(ctx context.Context, correlationId string, ...) error

+ func (c *MessageQueue) SendAsObject(ctx context.Context, correlationId string, messageType string, message any) (err error)

+ func (c *MessageQueue) SetReferences(ctx context.Context, references cref.IReferences)

+ func (c *MessageQueue) String() string

+ type MessagingCapabilities struct

+ func NewMessagingCapabilities(canMessageCount bool, canSend bool, canReceive bool, canPeek bool, ...) *MessagingCapabilities

+ func (c *MessagingCapabilities) CanAbandon() bool

+ func (c *MessagingCapabilities) CanClear() bool

+ func (c *MessagingCapabilities) CanDeadLetter() bool

+ func (c *MessagingCapabilities) CanMessageCount() bool

+ func (c *MessagingCapabilities) CanPeek() bool

+ func (c *MessagingCapabilities) CanPeekBatch() bool

+ func (c *MessagingCapabilities) CanReceive() bool

+ func (c *MessagingCapabilities) CanRenewLock() bool

+ func (c *MessagingCapabilities) CanSend() bool