mossdb

README

MossDB

MossDB is a in-memory, persistent and embedded key-value store.

Features:

• Key-value memory store, just use a simple map to store key/value
• Transactions supported
• AOL persistent, use the wal to store every commit, just like redis append-only file
• TTL supported, user can set the key with expire time that based on time heap
• Watch interface, caller will receive event when the value of key changes
• Embedded with a simple API
• Multi storage engine, now it support Map and RadixTree as backend storage

The architecture of MossDB as follows:

Getting Started

Getting MossDB

To start using MossDB, just use go get:

go get github.com/qingwave/mossdb

Using MossDB

There is a simple example shows how to use MossDB.

package main

import (
	"context"
	"errors"
	"log"
	"time"

	"github.com/qingwave/mossdb"
)

func () {
    // create db instance
	db, err := mossdb.New(&mossdb.Config{})
	if err != nil {
		log.Fatal(err)
	}

    // set, get, delete data
	db.Set("key1", []byte("val1"))
    log.Printf("get key1: %s", db.Get("key1"))
    db.Delete("key1", []byte("val1"))
}

Transactions

The database allows for rich transactions, in which multiple objects are inserted, updated or deleted. Note that one transaction allowed at a time.

    db.Tx(func(tx *mossdb.Tx) error {
        val1 := tx.Get("key1")

        tx.Set("key2", val1)

        return nil
    })

Watch

Watch interface just like etcd Watch(watch a key or prefix key), caller will receive event when the watched keys modified.

func Watch(db *mossdb.DB) {
	key := "watch-key"

	go func() {
		db.Set(key, mossdb.Val("val1"))
		db.Set(key, mossdb.Val("val2"))
		db.Delete(key)

		time.Sleep(100 * time.Second)
		db.Set(key, mossdb.Val("val3"))
	}()

	ctx, cancel := context.WithTimeout(context.Background(), 100*time.Millisecond)
	defer cancel()

	ch := db.Watch(ctx, key)

	for {
		select {
		case <-ctx.Done():
			log.Printf("context done")
			return
		case resp, ok := <-ch:
			if !ok {
				log.Printf("watch done")
				return
			}
			log.Printf("receive event: %s, key: %s, new val: %s", resp.Event.Op, resp.Event.Key, resp.Event.Val)
		}
	}
}

TTL

Key with expired time in useful sometimes, MossDB use a time heap to expire key near real-time.

db.Set("key-ttl", []byte("val-ttl"), mossdb.WithTTL(100*time.Millisecond))

Documentation

Index

• Constants
• Variables
• type Config
  • func (c *Config) Default() *Config
• type DB
  • func New(conf *Config) (*DB, error)
  • func (db *DB) Close()
  • func (db *DB) Delete(key string, opts ...Option) error
  • func (db *DB) Flush() error
  • func (db *DB) Get(key string, opts ...Option) Val
  • func (db *DB) List(opts ...Option) map[string][]byte
  • func (db *DB) Set(key string, val Val, opts ...Option) error
  • func (db *DB) Tx(f func(tx *Tx) error) error
  • func (db *DB) Watch(ctx context.Context, key string, opts ...Option) <-chan WatchResponse
• type Op
  • func NewOption(op OpType, key string, val Val, opts ...Option) *Op
  • func OpFromRecord(record *Record) *Op
  • func (op *Op) IsGet() bool
  • func (op *Op) IsMutate() bool
• type OpType
  • func (op OpType) String() string
• type Option
  • func WithAll() Option
  • func WithCreateNotify() Option
  • func WithDeleteNotify() Option
  • func WithMsg(msg string) Option
  • func WithPrefix() Option
  • func WithPrefixKey(key string) Option
  • func WithTTL(ttl time.Duration) Option
  • func WithUpdateNotify() Option
  • func WithWatchEventBuffSize(size int) Option
• type Record
  • func Decode(data []byte) (*Record, error)
  • func NewRecord(opt *Op) *Record
  • func (r *Record) Encode() ([]byte, error)
  • func (r *Record) Size() uint32
• type SubWatcher
  • func (sw *SubWatcher) PushEvent(e *WatchEvent)
• type Tx
  • func (tx *Tx) Delete(key string, opts ...Option)
  • func (tx *Tx) Get(key string, opts ...Option) Val
  • func (tx *Tx) List(opts ...Option) map[string][]byte
  • func (tx *Tx) Set(key string, val Val, opts ...Option)
• type Val
• type WatchEvent
• type WatchResponse
• type Watcher
  • func NewWatcher() *Watcher
  • func (w *Watcher) AddEvent(event *WatchEvent)
  • func (w *Watcher) Close()
  • func (w *Watcher) Run()
  • func (w *Watcher) Watch(ctx context.Context, key string, opts ...Option) <-chan WatchResponse

Constants

const DefaultWatcherQueueSize = 1024

const RecordHeaderSize = 2 + 4 + 4 + 8 + 8

Variables

var (
	ErrInvalidRecord = errors.New("invalid record")
)

Types

type Config

type Config struct {
	Path  string
	Store store.Store
}

func (*Config) Default

func (c *Config) Default() *Config

type DB

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

func New

func New(conf *Config) (*DB, error)

func (*DB) Close

func (db *DB) Close()

func (*DB) Delete

func (db *DB) Delete(key string, opts ...Option) error

func (*DB) Flush

func (db *DB) Flush() error

func (*DB) Get

func (db *DB) Get(key string, opts ...Option) Val

func (*DB) List

func (db *DB) List(opts ...Option) map[string][]byte

func (*DB) Set

func (db *DB) Set(key string, val Val, opts ...Option) error

func (*DB) Tx

func (db *DB) Tx(f func(tx *Tx) error) error

func (*DB) Watch

func (db *DB) Watch(ctx context.Context, key string, opts ...Option) <-chan WatchResponse

type Op

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

func NewOption

func NewOption(op OpType, key string, val Val, opts ...Option) *Op

func OpFromRecord

func OpFromRecord(record *Record) *Op

func (*Op) IsGet

func (op *Op) IsGet() bool

func (*Op) IsMutate

func (op *Op) IsMutate() bool

type OpType

type OpType uint16

const (
	GetOp OpType = iota + 1
	ModifyOp
	DeleteOp
	WatchOP
	ListOp
)

func (OpType) String

func (op OpType) String() string

type Option

type Option func(*Op)

func WithAll

func WithAll() Option

func WithCreateNotify

func WithCreateNotify() Option

func WithDeleteNotify

func WithDeleteNotify() Option

func WithMsg

func WithMsg(msg string) Option

func WithPrefix

func WithPrefix() Option

func WithPrefixKey

func WithPrefixKey(key string) Option

func WithTTL

func WithTTL(ttl time.Duration) Option

func WithUpdateNotify

func WithUpdateNotify() Option

func WithWatchEventBuffSize

func WithWatchEventBuffSize(size int) Option

type Record

type Record struct {
	Op        uint16
	KeySize   uint32
	ValSize   uint32
	Timestamp uint64
	TTL       uint64
	Key       []byte
	Val       []byte
}

func Decode

func Decode(data []byte) (*Record, error)

func NewRecord

func NewRecord(opt *Op) *Record

func (*Record) Encode

func (r *Record) Encode() ([]byte, error)

func (*Record) Size

func (r *Record) Size() uint32

type SubWatcher

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

func (*SubWatcher) PushEvent

func (sw *SubWatcher) PushEvent(e *WatchEvent)

type Tx

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

func (*Tx) Delete

func (tx *Tx) Delete(key string, opts ...Option)

func (*Tx) Get

func (tx *Tx) Get(key string, opts ...Option) Val

func (*Tx) List

func (tx *Tx) List(opts ...Option) map[string][]byte

func (*Tx) Set

func (tx *Tx) Set(key string, val Val, opts ...Option)

type Val

type Val []byte

type WatchEvent

type WatchEvent struct {
	Key    string
	Val    Val
	OldVal Val
	Op     OpType
}

type WatchResponse

type WatchResponse struct {
	Wid      string
	Event    *WatchEvent
	Canceled bool
}

type Watcher

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

func NewWatcher

func NewWatcher() *Watcher

func (*Watcher) AddEvent

func (w *Watcher) AddEvent(event *WatchEvent)

func (*Watcher) Close

func (w *Watcher) Close()

func (*Watcher) Run

func (w *Watcher) Run()

func (*Watcher) Watch

func (w *Watcher) Watch(ctx context.Context, key string, opts ...Option) <-chan WatchResponse