ppmq

type AckPair struct {
	Key     MessageKey
	Expires time.Time
}

type AckStorageEngine interface {
	// LoadAck loads slice of AckPair for given subscriber
	LoadAck(cqName string) ([]AckPair, error)
	// SaveAck saves slice of AckPair for given subscriber
	SaveAck(cqName string, waiting []AckPair) error
}

type AckWrapper struct {
	Ack []AckPair
}

type Application struct {
	Clock clock.Clock
	Conf  Config
	Test  bool
}

type Config struct {
	ServerSocket string         `yaml:"ServerSocket"` // socket for MQ app, to send commands, messages & ack
	DataPath     string         `yaml:"DataPath"`     // parent directory to store messages/queue/ack for each message topic
	Subscribers  []Subscriber   `yaml:"Subscribers"`  // list of subscribers
	Topics       []MessageTopic `yaml:"Topics"`       // list of message topics
	// contains filtered or unexported fields
}

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

type EngineImpl struct {
	Topic    MessageTopic                // Topic for which we are listening
	Messages PPMessageStore              // Messages is a store implementation for messages
	SQ       map[string]*SubscriberQueue // SQ is a map of subscriber name and SubscriberQueue pairs
	// contains filtered or unexported fields
}

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

type MQWrapper struct {
	// Topic name of the MessageTopic for which message is sent
	Topic string `json:"topic,omitempty"`
	// Subscriber name of the subscriber sending or receiving
	Subscriber string `json:"subscriber,omitempty"`
	// Messages slice of Message (used to send to Subscriber)
	Messages []Message `json:"messages,omitempty"`
	// Acks slice of MessageKey received from Subscriber to acknowledge processing
	Acks []MessageKey `json:"acks,omitempty"`
	// Command is sent by Subscriber, right now only on reconnect/restart of the Subscriber to reset Acks
	Command string `json:"command,omitempty""`
}

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

type MemoryMessageStore struct {
	Messages map[MessageKey]Message
	Meta     map[MessageKey]MessageMeta
}

type MemoryQueue struct {
	Queue []MessageKey
}

type Message struct {
	// MessageKey used for dedupe, ignored if dedupe in MessageTopic  is disabled
	Key MessageKey `json:",omitempty"`

	// MessagePayload the actual payload
	Payload MessagePayload

	// Expires time when message becomes expired, and should be thrown out, can be empty to represent messages
	// that don't expire
	Expires time.Time `json:",omitempty"`

	// Priority defines message priority, starts with 0 (highest priority)
	Priority int `json:",omitempty"`
}

type MessageKey string

type MessageMeta struct {
	AckCount int
	Sent     map[string]bool
}

type MessagePayload string

type MessageTopic struct {
	Name     string        `yaml:"Name"`               // name of the topic
	AckRetry time.Duration `yaml:"AckRetry,omitempty"` //retry resending message if Ack was received in
	// AckRetry interval, 0 if retry disabled
	Dedupe bool `yaml:"Dedupe,omitempty"` // if true, messages will be dedupe based on MessageKey,
	// latest will be always used instead of old message
	Priorities int      `yaml:"Priorities,omitempty"` // 1 or more, if 1 - no priorities...
	Expire     bool     `yaml:"Expire,omitempty"`     // True if message can expire
	Subscriber []string `yaml:"Subscribers"`          // List of subscribers
	// contains filtered or unexported fields
}

type PPMessageStore interface {
	// Init message store engine
	Init() error
	// Close message store engine, safely persisting all messages if needed
	Close() error
	// Get Message and MessageMeta for given MessageKey. Returns ErrNoSuchMessage if doesn't exist
	Get(key MessageKey) (Message, MessageMeta, error)
	// GetMeta returns MessageMeta for given MessageKey, Returns ErrNoSuchMessage if doesn't exist
	GetMeta(key MessageKey) (MessageMeta, error)
	// Put persists Message & MessageMeta
	Put(m Message, meta MessageMeta) error
	// UpdateMeta updates MessageMeta
	UpdateMeta(key MessageKey, meta MessageMeta) error
	// Delete removes Message & MessageMeta for given key from the storage
	Delete(key MessageKey) error
	// Exists checks if Message exists
	Exists(key MessageKey) (bool, error)
	// Sync saves all changes to disk
	Sync() error
	// Abort tries to safely abort the operation. Not implemented at this moment.
	Abort() error
}

type PPQueue interface {
	// Init underlying persistence layer
	Init() error
	// Push message key into the queue
	Push(key MessageKey) error
	// Prepend slice of message keys into the queue. Used on ack reset & timeout, when we want to resend some messages
	Prepend(keys []MessageKey) error
	// Pop message from the queue
	Pop() (MessageKey, error)
	// Peek check first available message on the queue
	Peek() (MessageKey, error)
	// IsEmpty returns true if queue is empty
	IsEmpty() (bool, error)
	// Sync saves queue state
	Sync() error
	// Abort attempts to abort the changes to the queue before saving (might not be implemented)
	Abort() error
	// Closes the queue, saving all unsaved data.
	Close() error
}

type PersistentFactory struct {
	Path string // location where data should be stored for given MessageTopic
	// contains filtered or unexported fields
}

type StorageEngine interface {
	// Init initializes StorageEngine for given MessageTopic, using particular StorageFactory
	Init(mq MessageTopic, factory StorageFactory) error
	// Closes all underlying persistence engines, making sure that data is saved (if supported)
	Close() error
	// Checks if the are available messages for given subscriber that can be served now
	IsAvailable(cqName string) (bool, error)
	// Retrieves up to count messages for given subscriber
	NextBatch(cqName string, count int) ([]Message, error)
	// Returns slice of subscribers (names) for given StorageEngine
	GetNamesOfSubscribers() []string
	// Add slice of messages to the MessageTopic
	Add(msgs []Message) error
	// Ack slice of messages for given subscriber
	Ack(cqName string, keys []MessageKey) error
	// ResetAck resets unacknowledged messages for given subscriber
	ResetAck(cqName string) error
	// TimeoutAck resets expired ack for all subscribers for MessageTopic
	TimeoutAck() error
	// TestState is used during testing process to test final state of the queue for a given subscriber
	// It should be assumed that the method is destructive, and the queue/MessageTopic will be in invalid state
	// afterwards. Used specifically for testing purposes.
	TestState(cqName string, queue []MessageKey, ack []AckPair) error
}

type StorageFactory interface {
	// MessageStore returns persistence engine for messages
	MessageStore(topicName string) (PPMessageStore, error)
	// MessageQueue returns persistence engine for queue
	MessageQueue(topicName, subscriberName string) (PPQueue, error)
	// AckStore returns persistence engine for Ack
	AckStore(topicName string) (AckStorageEngine, error)
}

type Subscriber struct {
	Id        int    `yaml:",omitempty"`          // subscriber id (generated on init)
	Name      string `yaml:"Name"`                // subscriber name
	BatchSize int    `yaml:"BatchSize,omitempty"` // max of messages to send to subscriber in one go
	Socket    string `yaml:"Socket"`              // socket to which to send the messages
}

type SubscriberQueue struct {
	Name       string           // Name of the subscriber
	Queue      []PPQueue        // Queue is a slice of queues, one per priority, highest starts with 0
	AckWaiting []AckPair        // AckWaiting list ack for message id
	AckEngine  AckStorageEngine // used to store ack pairs
}