Documentation
¶
Index ¶
- Constants
- func NewPostgresMessageQueue(pgConn storage.PostgresConfig, name string, interval time.Duration, ...) (convdomain.MessageQueue, error)
- func NewRedisMessageQueue(conn storage.RedisConfig, name string, interval time.Duration, wake func()) (convdomain.MessageQueue, error)
- func NewSQLiteMessageQueue(sqliteConn storage.SQLiteConfig, name string, interval time.Duration, ...) (convdomain.MessageQueue, error)
- type MessageQueueService
- func (mq *MessageQueueService) BackendName() string
- func (mq *MessageQueueService) Clear(_ context.Context) error
- func (mq *MessageQueueService) Dequeue(_ context.Context) (*convdomain.QueuedMessage, error)
- func (mq *MessageQueueService) Drain(_ context.Context) ([]convdomain.QueuedMessage, error)
- func (mq *MessageQueueService) Enqueue(_ context.Context, message sdk.Message, source convdomain.QueuedMessageSource, ...) error
- func (mq *MessageQueueService) GetAll() []convdomain.QueuedMessage
- func (mq *MessageQueueService) IsEmpty() bool
- func (mq *MessageQueueService) Size() int
- func (mq *MessageQueueService) Unreachable() bool
Constants ¶
const RedisQueueKeyPrefix = "infer:queue:"
RedisQueueKeyPrefix namespaces the queue lists. External producers feed a running infer by pushing a JSON queuedRecord to infer:queue:<name> (RPUSH).
Variables ¶
This section is empty.
Functions ¶
func NewPostgresMessageQueue ¶
func NewPostgresMessageQueue(pgConn storage.PostgresConfig, name string, interval time.Duration, wake func()) (convdomain.MessageQueue, error)
NewPostgresMessageQueue dials the storage.postgres connection, applies the schema migrations and returns the durable queue adapter. Startup fails fast with an error naming the backend when PostgreSQL is unreachable.
func NewRedisMessageQueue ¶
func NewRedisMessageQueue(conn storage.RedisConfig, name string, interval time.Duration, wake func()) (convdomain.MessageQueue, error)
NewRedisMessageQueue dials the storage.redis connection and returns the Redis-backed queue adapter. Startup fails fast with an error naming the backend when Redis is unreachable.
func NewSQLiteMessageQueue ¶
func NewSQLiteMessageQueue(sqliteConn storage.SQLiteConfig, name string, interval time.Duration, wake func()) (convdomain.MessageQueue, error)
NewSQLiteMessageQueue opens the storage.sqlite database, applies the schema migrations and returns the durable queue adapter. Startup fails fast with an error naming the backend when the database file is unusable.
Types ¶
type MessageQueueService ¶
type MessageQueueService struct {
// contains filtered or unexported fields
}
MessageQueueService manages the in-memory message queue: the process-local default backend where nothing can fail and nothing survives a restart.
func NewMessageQueueService ¶
func NewMessageQueueService() *MessageQueueService
NewMessageQueueService creates a new message queue service
func (*MessageQueueService) BackendName ¶
func (mq *MessageQueueService) BackendName() string
BackendName names the active backend for the status bar indicator
func (*MessageQueueService) Clear ¶
func (mq *MessageQueueService) Clear(_ context.Context) error
Clear removes all messages from the queue
func (*MessageQueueService) Dequeue ¶
func (mq *MessageQueueService) Dequeue(_ context.Context) (*convdomain.QueuedMessage, error)
Dequeue removes and returns the next message from the queue Returns nil if the queue is empty
func (*MessageQueueService) Drain ¶
func (mq *MessageQueueService) Drain(_ context.Context) ([]convdomain.QueuedMessage, error)
Drain removes and returns every queued message, oldest first
func (*MessageQueueService) Enqueue ¶
func (mq *MessageQueueService) Enqueue(_ context.Context, message sdk.Message, source convdomain.QueuedMessageSource, requestID string) error
Enqueue adds a message to the queue with its origin source and request ID
func (*MessageQueueService) GetAll ¶
func (mq *MessageQueueService) GetAll() []convdomain.QueuedMessage
GetAll returns all messages in the queue without removing them
func (*MessageQueueService) IsEmpty ¶
func (mq *MessageQueueService) IsEmpty() bool
IsEmpty returns true if the queue has no messages
func (*MessageQueueService) Size ¶
func (mq *MessageQueueService) Size() int
Size returns the number of messages in the queue
func (*MessageQueueService) Unreachable ¶
func (mq *MessageQueueService) Unreachable() bool
Unreachable reports whether the backend failed its latest operation; the memory queue never fails
Source Files