queue

type Backend interface {
	// Backend type.
	Type() string
	// Ping connect the backend server if not connected.
	// Will be called before every Read/Write operation.
	Ping() error
	// Return the max delay duration supported by the backend.
	// A negative value means no limitation.
	// A zero value means delay operation is not supported.
	MaxDelay() time.Duration
	// Return all queue names in backend storage.
	GetQueues() ([]string, error)
	// Return all queue/topics in backend storage.
	GetTopics() (map[string][]string, error)
	// Return all topic length of specified queue in backend storage.
	GetQueueLength(queue string) (map[string]int64, error)
	// Return the specified queue/topic length in backend storage.
	GetTopicLength(queue, topic string) (int64, error)

	// Read subscribes the message of the specified queue and topic.
	Read(ctx context.Context, queue, topic string, ch chan<- MessageWrapper) error
	// Write publishes content data to the specified queue.
	Write(ctx context.Context, queue string, delay time.Duration, content []byte) error
}

type Consumer interface {
	Handle(context.Context, Message) error
}

type ConsumerFunc func(context.Context, Message) error

type Idempotent interface {
	// Invoked before process message.
	// Returns true to continue the message processing.
	// Returns false to invoke Cancel for the message.
	BeforeProcess(Message) bool
	// Invoked after processing.
	AfterProcess(Message, ProcessStatus)
}

type Message interface {
	// Queue name of this message.
	Queue() string
	// Topic name of this message.
	Topic() string

	// Unique ID of this message.
	UniqueID() string
	// Message body content.
	Content() []byte
	// The creation time of the message.
	Timestamp() time.Time
	// The message should not be processed before this timestamp.
	NotBefore() time.Time
	// Message retry times.
	Retry() int
	// Return true for a ping message.
	IsPing() bool
}

type MessageOperation interface {
	// Begin to process the message.
	Begin()
	// Indicate the message should be ignored.
	Cancel()
	// End indicates a successful process.
	End()
	// Requeue indicates the message should be retried.
	Requeue()
	// Fail indicates a failed process.
	Fail()
}

type MessageWrapper interface {
	Message
	MessageOperation
}

type PingMessage struct{}

type ProcessStatus int

type PublishOption func(*PublishOptions)

type PublishOptions struct {
	context.Context
	Delay time.Duration
}

type Queue interface {
	// Publish writes a message body to the specified queue.
	Publish(queue string, content []byte, opts ...PublishOption) error
	// Subscribe consumes the messages of the specified queue.
	Subscribe(queue string, handler Consumer, opts ...SubscribeOption) error
}

type SubscribeOption func(*SubscribeOptions)

type SubscribeOptions struct {
	context.Context
	Component   string
	Product     string
	Topic       string
	Concurrency int
	MaxRetry    int
	Idempotent  Idempotent
}