msgstore

Documentation

Index

• Variables
• type IndexMessageFactory
  • func (imf *IndexMessageFactory) New() proto.Unmarshaler
• type MessageContentFactory
  • func (mcf *MessageContentFactory) New() proto.Unmarshaler
• type MessageStore
  • func NewMessageStore(fileName string) (*MessageStore, error)
  • func (ms *MessageStore) AddMessage(msg *amqp.Message, queues []string) (map[string][]*amqp.QueueMessage, error)
  • func (ms *MessageStore) AddTxMessages(msgs []*amqp.TxMessage) (map[string][]*amqp.QueueMessage, error)
  • func (ms *MessageStore) Fsck() ([]int64, []int64)
  • func (ms *MessageStore) Get(qm *amqp.QueueMessage, rhs []amqp.MessageResourceHolder) (*amqp.Message, bool)
  • func (ms *MessageStore) GetAndDecrRef(qm *amqp.QueueMessage, queueName string, rhs []amqp.MessageResourceHolder) (*amqp.Message, error)
  • func (ms *MessageStore) GetIndex(id int64) (msg *amqp.IndexMessage, found bool)
  • func (ms *MessageStore) GetNoChecks(id int64) (msg *amqp.Message, found bool)
  • func (ms *MessageStore) IncrDeliveryCount(queueName string, qm *amqp.QueueMessage) (err error)
  • func (ms *MessageStore) IndexCount() int
  • func (ms *MessageStore) LoadMessages() error
  • func (ms *MessageStore) LoadQueueFromDisk(queueName string) (*list.List, error)
  • func (ms *MessageStore) MessageCount() int
  • func (ms *MessageStore) RemoveRef(qm *amqp.QueueMessage, queueName string, rhs []amqp.MessageResourceHolder) error
  • func (ms *MessageStore) Start()
• type PersistKey
• type QueueMessageFactory
  • func (qmf *QueueMessageFactory) New() proto.Unmarshaler
• type TestResourceHolder
  • func (trh *TestResourceHolder) AcquireResources(qm *amqp.QueueMessage) bool
  • func (trh *TestResourceHolder) ReleaseResources(qm *amqp.QueueMessage)

Variables

var MESSAGE_CONTENT_BUCKET = []byte("message_content")

var MESSAGE_INDEX_BUCKET = []byte("message_index")

Types

type IndexMessageFactory

type IndexMessageFactory struct{}

func (*IndexMessageFactory) New

func (imf *IndexMessageFactory) New() proto.Unmarshaler

type MessageContentFactory

type MessageContentFactory struct{}

func (*MessageContentFactory) New

func (mcf *MessageContentFactory) New() proto.Unmarshaler

type MessageStore

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

func NewMessageStore

func NewMessageStore(fileName string) (*MessageStore, error)

func (*MessageStore) AddMessage

func (ms *MessageStore) AddMessage(msg *amqp.Message, queues []string) (map[string][]*amqp.QueueMessage, error)

func (*MessageStore) AddTxMessages

func (ms *MessageStore) AddTxMessages(msgs []*amqp.TxMessage) (map[string][]*amqp.QueueMessage, error)

func (*MessageStore) Fsck

func (ms *MessageStore) Fsck() ([]int64, []int64)

func (*MessageStore) Get

func (ms *MessageStore) Get(qm *amqp.QueueMessage, rhs []amqp.MessageResourceHolder) (*amqp.Message, bool)

func (*MessageStore) GetAndDecrRef

func (ms *MessageStore) GetAndDecrRef(qm *amqp.QueueMessage, queueName string, rhs []amqp.MessageResourceHolder) (*amqp.Message, error)

func (*MessageStore) GetIndex

func (ms *MessageStore) GetIndex(id int64) (msg *amqp.IndexMessage, found bool)

func (*MessageStore) GetNoChecks

func (ms *MessageStore) GetNoChecks(id int64) (msg *amqp.Message, found bool)

func (*MessageStore) IncrDeliveryCount

func (ms *MessageStore) IncrDeliveryCount(queueName string, qm *amqp.QueueMessage) (err error)

func (*MessageStore) IndexCount

func (ms *MessageStore) IndexCount() int

func (*MessageStore) LoadMessages

func (ms *MessageStore) LoadMessages() error

func (*MessageStore) LoadQueueFromDisk

func (ms *MessageStore) LoadQueueFromDisk(queueName string) (*list.List, error)

func (*MessageStore) MessageCount

func (ms *MessageStore) MessageCount() int

func (*MessageStore) RemoveRef

func (ms *MessageStore) RemoveRef(qm *amqp.QueueMessage, queueName string, rhs []amqp.MessageResourceHolder) error

func (*MessageStore) Start

func (ms *MessageStore) Start()

type PersistKey

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

type QueueMessageFactory

type QueueMessageFactory struct{}

func (*QueueMessageFactory) New

func (qmf *QueueMessageFactory) New() proto.Unmarshaler

type TestResourceHolder

type TestResourceHolder struct {
}

func (*TestResourceHolder) AcquireResources

func (trh *TestResourceHolder) AcquireResources(qm *amqp.QueueMessage) bool

func (*TestResourceHolder) ReleaseResources

func (trh *TestResourceHolder) ReleaseResources(qm *amqp.QueueMessage)