pubsub

Documentation

Index

• Variables
• func DecodeInt64(b []byte) int64
• func EncodeInt64(v int64) []byte
• func MessageKey(topic *Topic, offset *Offset) []byte
• func SnapshotKey(t *Topic, s *Subscription, name string) []byte
• func SubscriptionKey(topic *Topic, sub string) []byte
• func TopicKey(name string) []byte
• func UUID() []byte
• type Message
• type MessageID
• type Offset
  • func OffsetFromBytes(d []byte) *Offset
  • func OffsetFromString(s string) *Offset
  • func (offset *Offset) Bytes() []byte
  • func (offset *Offset) Next() *Offset
  • func (offset *Offset) String() string
• type Pubsub
  • func MockOpen(path string) (*Pubsub, error)
  • func Open(path string) (*Pubsub, error)
  • func (p *Pubsub) Begin() (*Transaction, error)
• type ScanHandler
• type Snapshot
• type Subscription
• type Topic
• type Transaction
  • func (txn *Transaction) Append(topic *Topic, messages ...*Message) ([]MessageID, error)
  • func (txn *Transaction) Commit(ctx context.Context) error
  • func (txn *Transaction) CreateSnapshot(topic *Topic, subscription *Subscription, name string) (*Snapshot, error)
  • func (txn *Transaction) CreateSubscription(t *Topic, name string) (*Subscription, error)
  • func (txn *Transaction) CreateTopic(name string) (*Topic, error)
  • func (txn *Transaction) DeleteSnapshot(topic *Topic, subscription *Subscription, name string) error
  • func (txn *Transaction) DeleteSubscription(t *Topic, name string) error
  • func (txn *Transaction) DeleteTopic(name string) error
  • func (txn *Transaction) GetSnapshot(topic *Topic, subscription *Subscription, name string) (*Snapshot, error)
  • func (txn *Transaction) GetSnapshots(topic *Topic, subscription *Subscription) ([]*Snapshot, error)
  • func (txn *Transaction) GetSubscription(t *Topic, name string) (*Subscription, error)
  • func (txn *Transaction) GetSubscriptions(t *Topic) ([]*Subscription, error)
  • func (txn *Transaction) GetTopic(name string) (*Topic, error)
  • func (txn *Transaction) Rollback() error
  • func (txn *Transaction) Scan(topic *Topic, offset *Offset, handler ScanHandler) error
  • func (txn *Transaction) UpdateSubscription(t *Topic, s *Subscription) error

Variables

var ErrNotFound = errors.New("not found")

Functions

func DecodeInt64

func DecodeInt64(b []byte) int64

DecodeInt64 decodes an int64

func EncodeInt64

func EncodeInt64(v int64) []byte

EncodeInt64 encodes an int64 to be memcomparable in asc order

func MessageKey

func MessageKey(topic *Topic, offset *Offset) []byte

MessageKey builds a key of a message

func SnapshotKey

func SnapshotKey(t *Topic, s *Subscription, name string) []byte

SnapshotKey builds a key of a snapshot

func SubscriptionKey

func SubscriptionKey(topic *Topic, sub string) []byte

SubscriptionKey builds a key of a subscription

func TopicKey

func TopicKey(name string) []byte

TopicKey builds a key of a topic

func UUID

func UUID() []byte

UUID generates a global unique ID

Types

type Message

type Message struct {
	Payload []byte
}

Message wraps a bytes payload

type MessageID

type MessageID struct {
	*Offset
}

MessageID is an identity of message

type Offset

type Offset struct {
	TS    int64 // TS is the StartTS of the transaction
	Index int64
}

Offset is the position of a message in a topic

func OffsetFromBytes

func OffsetFromBytes(d []byte) *Offset

OffsetFromBytes parses offset from bytes

func OffsetFromString

func OffsetFromString(s string) *Offset

OffsetFromString parses offset from a string

func (*Offset) Bytes

func (offset *Offset) Bytes() []byte

Bytes returns offset as bytes

func (*Offset) Next

func (offset *Offset) Next() *Offset

Next returns a greater offset

func (*Offset) String

func (offset *Offset) String() string

String returns offset as human-friendly string

type Pubsub

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

Pubsub is a storage with a pub/sub interface

func MockOpen

func MockOpen(path string) (*Pubsub, error)

MockOpen open a faked pubsub storage

func Open

func Open(path string) (*Pubsub, error)

Open a pubsub storage

func (*Pubsub) Begin

func (p *Pubsub) Begin() (*Transaction, error)

Begin a transaction

type ScanHandler

type ScanHandler func(id MessageID, message *Message) bool

ScanHandler is a handler to process scanned messages

type Snapshot

type Snapshot struct {
	Name         string
	Subscription *Subscription
}

Snapshot is an immutable state of a subscription

type Subscription

type Subscription struct {
	Name  string
	Sent  *Offset
	Acked *Offset
}

Subscription keeps the state of subscribers

type Topic

type Topic struct {
	Name      string
	ObjectID  []byte
	CreatedAt int64
}

Topic is the meta of a topic

type Transaction

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

Transaction suppies the api to access pubsub

func (*Transaction) Append

func (txn *Transaction) Append(topic *Topic, messages ...*Message) ([]MessageID, error)

Append a message to a topic

func (*Transaction) Commit

func (txn *Transaction) Commit(ctx context.Context) error

Commit a transaction

func (*Transaction) CreateSnapshot

func (txn *Transaction) CreateSnapshot(topic *Topic, subscription *Subscription, name string) (*Snapshot, error)

CreateSnapshot creates a snapshot for a subscription

func (*Transaction) CreateSubscription

func (txn *Transaction) CreateSubscription(t *Topic, name string) (*Subscription, error)

CreateSubscritpion creates a subscription

func (*Transaction) CreateTopic

func (txn *Transaction) CreateTopic(name string) (*Topic, error)

CreateTopic creates a topic, if the topic has existed, return it

func (*Transaction) DeleteSnapshot

func (txn *Transaction) DeleteSnapshot(topic *Topic, subscription *Subscription, name string) error

DeleteSnapshot deletes a snapshot

func (*Transaction) DeleteSubscription

func (txn *Transaction) DeleteSubscription(t *Topic, name string) error

DeleteSubscription deletes a subscription

func (*Transaction) DeleteTopic

func (txn *Transaction) DeleteTopic(name string) error

DeleteTopic deletes a topic

func (*Transaction) GetSnapshot

func (txn *Transaction) GetSnapshot(topic *Topic, subscription *Subscription, name string) (*Snapshot, error)

GetSnapshot returns a snapshot

func (*Transaction) GetSnapshots

func (txn *Transaction) GetSnapshots(topic *Topic, subscription *Subscription) ([]*Snapshot, error)

GetSnapshots lists all snapshots of a subscription

func (*Transaction) GetSubscription

func (txn *Transaction) GetSubscription(t *Topic, name string) (*Subscription, error)

GetSubscription returns a subscription

func (*Transaction) GetSubscriptions

func (txn *Transaction) GetSubscriptions(t *Topic) ([]*Subscription, error)

GetSubscriptions lists all subscriptions of a topic

func (*Transaction) GetTopic

func (txn *Transaction) GetTopic(name string) (*Topic, error)

GetTopic gets the topic information

func (*Transaction) Rollback

func (txn *Transaction) Rollback() error

Rollback a transaction

func (*Transaction) Scan

func (txn *Transaction) Scan(topic *Topic, offset *Offset, handler ScanHandler) error

Scan seeks to the offset and calls handler for each message

func (*Transaction) UpdateSubscription

func (txn *Transaction) UpdateSubscription(t *Topic, s *Subscription) error

UpdateSubscription updates a subscription