journal

Documentation

Index

• Constants
• Variables
• type Add
• type Constraint
• type Delete
  • func (d *Delete) Key() string
• type Entry
• type ID
• type Queue
  • func Create(ctx context.Context, engine storage.Engine, path *storage.URI, base ID) (*Queue, error)
  • func New(engine storage.Engine, path *storage.URI) *Queue
  • func Open(ctx context.Context, engine storage.Engine, path *storage.URI) (*Queue, error)
  • func (q *Queue) Boundaries(ctx context.Context) (ID, ID, error)
  • func (q *Queue) Commit(ctx context.Context, b []byte) (ID, error)
  • func (q *Queue) CommitAt(ctx context.Context, at ID, b []byte) error
  • func (q *Queue) Load(ctx context.Context, id ID) ([]byte, error)
  • func (q *Queue) MoveTail(ctx context.Context, id, base ID) error
  • func (q *Queue) NewReader(ctx context.Context, head, tail ID) *Reader
  • func (q *Queue) Open(ctx context.Context, head, tail ID) (io.Reader, error)
  • func (q *Queue) OpenAsZNG(ctx context.Context, zctx *zed.Context, head, tail ID) (*zngio.Reader, error)
  • func (q *Queue) Path() *storage.URI
  • func (q *Queue) ReadHead(ctx context.Context) (ID, error)
  • func (q *Queue) ReadTail(ctx context.Context) (ID, ID, error)
• type Reader
  • func (r *Reader) Read(b []byte) (int, error)
• type Store
  • func CreateStore(ctx context.Context, engine storage.Engine, logger *zap.Logger, ...) (*Store, error)
  • func OpenStore(ctx context.Context, engine storage.Engine, logger *zap.Logger, ...) (*Store, error)
  • func (s *Store) All(ctx context.Context) ([]Entry, error)
  • func (s *Store) Delete(ctx context.Context, key string, c Constraint) error
  • func (s *Store) Insert(ctx context.Context, e Entry) error
  • func (s *Store) Keys(ctx context.Context, key string) ([]string, error)
  • func (s *Store) Lookup(ctx context.Context, key string) (Entry, error)
  • func (s *Store) Move(ctx context.Context, oldKey string, newEntry Entry) error
  • func (s *Store) Update(ctx context.Context, e Entry, c Constraint) error
  • func (s *Store) Values(ctx context.Context) ([]interface{}, error)
• type Update

Constants

const MaxReadRetry = 10

Variables

var (
	ErrEmpty  = errors.New("empty log")
	ErrFailed = errors.New("transaction failed")
)

var (
	ErrRetriesExceeded = fmt.Errorf("config journal unavailable after %d attempts", maxRetries)
	ErrKeyExists       = errors.New("key already exists")
	ErrNoSuchKey       = errors.New("no such key")
	ErrConstraint      = errors.New("constraint failed")
)

Types

type Add

type Add struct {
	Entry `zed:"entry"`
}

type Constraint

type Constraint func(Entry) bool

type Delete

type Delete struct {
	EntryKey string `zed:"entry_key"`
}

func (*Delete) Key

func (d *Delete) Key() string

type Entry

type Entry interface {
	Key() string
}

type ID

type ID uint64

const Nil ID = 0

type Queue

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

func Create

func Create(ctx context.Context, engine storage.Engine, path *storage.URI, base ID) (*Queue, error)

func New

func New(engine storage.Engine, path *storage.URI) *Queue

func Open

func Open(ctx context.Context, engine storage.Engine, path *storage.URI) (*Queue, error)

func (*Queue) Boundaries

func (q *Queue) Boundaries(ctx context.Context) (ID, ID, error)

func (*Queue) Commit

func (q *Queue) Commit(ctx context.Context, b []byte) (ID, error)

XXX This needs concurrency work. See issue #2546.

func (*Queue) CommitAt

func (q *Queue) CommitAt(ctx context.Context, at ID, b []byte) error

CommitAt commits a new serialized ZNG sequence to the journal presuming the previous state conformed to the journal position "at". The entry is written at the next position in the log if possible. Otherwise, a write conflict occurs and an error is returned.

func (*Queue) Load

func (q *Queue) Load(ctx context.Context, id ID) ([]byte, error)

func (*Queue) MoveTail

func (q *Queue) MoveTail(ctx context.Context, id, base ID) error

MoveTail moves the tail of the journal to the indicated ID and does no validation. Use with caution. This update must be made by an exclusive write-lock that is outside the scope of the journal package. Unlike HEAD, TAIL is not a hint and must be consistent with the actual log entries at all times.

func (*Queue) NewReader

func (q *Queue) NewReader(ctx context.Context, head, tail ID) *Reader

NewReader returns a zngio.Reader that concatenates the journal files in sequence from tail to head. Since ZNG is stored in the journal, this produce a byte stream suitable for wrapper in a zngio.Reader.

func (*Queue) Open

func (q *Queue) Open(ctx context.Context, head, tail ID) (io.Reader, error)

func (*Queue) OpenAsZNG

func (q *Queue) OpenAsZNG(ctx context.Context, zctx *zed.Context, head, tail ID) (*zngio.Reader, error)

func (*Queue) Path

func (q *Queue) Path() *storage.URI

func (*Queue) ReadHead

func (q *Queue) ReadHead(ctx context.Context) (ID, error)

func (*Queue) ReadTail

func (q *Queue) ReadTail(ctx context.Context) (ID, ID, error)

type Reader

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

func (*Reader) Read

func (r *Reader) Read(b []byte) (int, error)

type Store

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

func CreateStore

func CreateStore(ctx context.Context, engine storage.Engine, logger *zap.Logger, path *storage.URI, keyTypes ...interface{}) (*Store, error)

func OpenStore

func OpenStore(ctx context.Context, engine storage.Engine, logger *zap.Logger, path *storage.URI, keyTypes ...interface{}) (*Store, error)

func (*Store) All

func (s *Store) All(ctx context.Context) ([]Entry, error)

func (*Store) Delete

func (s *Store) Delete(ctx context.Context, key string, c Constraint) error

func (*Store) Insert

func (s *Store) Insert(ctx context.Context, e Entry) error

func (*Store) Keys

func (s *Store) Keys(ctx context.Context, key string) ([]string, error)

func (*Store) Lookup

func (s *Store) Lookup(ctx context.Context, key string) (Entry, error)

func (*Store) Move

func (s *Store) Move(ctx context.Context, oldKey string, newEntry Entry) error

func (*Store) Update

func (s *Store) Update(ctx context.Context, e Entry, c Constraint) error

func (*Store) Values

func (s *Store) Values(ctx context.Context) ([]interface{}, error)

type Update

type Update struct {
	Entry `zed:"entry"`
}