store

Documentation

Index

• Variables
• func Boot(store broker.TopicMessageStore, event broker.MessageStoreEvent)
• func DeleteTopicMessageID(ctx context.Context, topic, messageID string) error
• func ReadPublishMessage(ctx context.Context, topic, startMessageID string, size int, include bool, ...) (err error)
• func ReadTopicWillMessage(ctx context.Context, topic, messageID string, ...) error
• type Local
  • func NewLocalStore(options nutsdb.Options, option ...nutsdb.Option) *Local
  • func (s *Local) CreatePacket(topic string, value []byte) (id string, err error)
  • func (s *Local) DeleteBeforeID(id string)
  • func (s *Local) DeleteTopicMessageID(ctx context.Context, topic, messageID string) error
  • func (s *Local) ReadFromTimestamp(ctx context.Context, topic string, timestamp time.Time, limit int) ([]packet.PublishMessage, error)
  • func (s *Local) ReadTopicMessagesByID(ctx context.Context, topic, id string, limit int, include bool) ([]packet.PublishMessage, error)
• type Wrapper
  • func NewStoreWrapper() *Wrapper
  • func (s *Wrapper) StorePublishPacket(topics map[string]int32, packet *packet2.PublishMessage) (messageID string, err error)

Variables

var (
	DefaultMessageStore      broker.TopicMessageStore
	DefaultMessageStoreEvent broker.MessageStoreEvent
	DefaultSerializerVersion serializer.SerialVersion
)

Functions

func Boot

func Boot(store broker.TopicMessageStore, event broker.MessageStoreEvent)

func DeleteTopicMessageID

func DeleteTopicMessageID(ctx context.Context, topic, messageID string) error

func ReadPublishMessage

func ReadPublishMessage(ctx context.Context, topic, startMessageID string, size int, include bool, writer func(message *packet2.PublishMessage)) (err error)

ReadPublishMessage reads the published message with the given topic and messageID from store if startMessageID is empty, read from the latest message if include is true, read from the startMessageID, otherwise read from the next message of startMessageID if include is true and startMessageID was the latest message, waiting for the next message by listening store event read message from store and write to writer

func ReadTopicWillMessage

func ReadTopicWillMessage(ctx context.Context, topic, messageID string, writer func(message *packet2.PublishMessage)) error

Types

type Local

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

func NewLocalStore

func NewLocalStore(options nutsdb.Options, option ...nutsdb.Option) *Local

func (*Local) CreatePacket

func (s *Local) CreatePacket(topic string, value []byte) (id string, err error)

func (*Local) DeleteBeforeID

func (s *Local) DeleteBeforeID(id string)

func (*Local) DeleteTopicMessageID

func (s *Local) DeleteTopicMessageID(ctx context.Context, topic, messageID string) error

func (*Local) ReadFromTimestamp

func (s *Local) ReadFromTimestamp(ctx context.Context, topic string, timestamp time.Time, limit int) ([]packet.PublishMessage, error)

func (*Local) ReadTopicMessagesByID

func (s *Local) ReadTopicMessagesByID(ctx context.Context, topic, id string, limit int, include bool) ([]packet.PublishMessage, error)

type Wrapper

type Wrapper struct {
}

Wrapper is a wrapper of pkg.Store

func NewStoreWrapper

func NewStoreWrapper() *Wrapper

func (*Wrapper) StorePublishPacket

func (s *Wrapper) StorePublishPacket(topics map[string]int32, packet *packet2.PublishMessage) (messageID string, err error)

StorePublishPacket stores the published packet to store if topics is empty, return error topics include the origin topic and the topic of the wildcard subscription and emit store event