pubsub

package

v0.0.0-...-b4d2924 Latest Latest Go to latest Published: Jun 4, 2019 License: Apache-2.0 Imports: 11 Imported by: 2

Details

Valid go.mod file

The Go module system was introduced in Go 1.11 and is the official dependency management solution for Go.
Redistributable license

Redistributable licenses place minimal restrictions on how software can be used, modified, and redistributed.
Tagged version

Modules with tagged versions give importers more predictable builds.
Stable version

When a project reaches major version v1 it is considered stable.
Learn more about best practices

Repository

github.com/tipsio/tips

Links

Open Source Insights

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

Constants ¶

This section is empty.

Variables ¶

View Source

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

Source Files ¶

View all Source files

?	: This menu
/	: Search site
f or F	: Jump to
y or Y	: Canonical URL