// Copyright © 2016 Abcum Ltd // // Licensed under the Apache License, Version 2.0 (the "License"); // you may not use this file except in compliance with the License. // You may obtain a copy of the License at // // http://www.apache.org/licenses/LICENSE-2.0 // // Unless required by applicable law or agreed to in writing, software // distributed under the License is distributed on an "AS IS" BASIS, // WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. // See the License for the specific language governing permissions and // limitations under the License. package kvs import ( "github.com/boltdb/bolt" ) // DB is a database handle to a single Surreal cluster. type DB struct { db *bolt.DB } func New(path string) (db *DB, err error) { var bo *bolt.DB bo, err = bolt.Open(path, 0666, nil) bo.Update(func(tx *bolt.Tx) error { tx.CreateBucketIfNotExists(bucket) return nil }) return &DB{db: bo}, err } func (db *DB) All() (kvs []*KV, err error) { tx, err := db.Txn(false) if err != nil { return } defer tx.Close() return tx.All() } // Get retrieves a single key:value item. func (db *DB) Get(key []byte) (kv *KV, err error) { tx, err := db.Txn(false) if err != nil { return } defer tx.Close() return tx.Get(key) } // MGet retrieves multiple key:value items. func (db *DB) MGet(keys ...[]byte) (kvs []*KV, err error) { tx, err := db.Txn(false) if err != nil { return } defer tx.Close() return tx.MGet(keys...) } // PGet retrieves the range of rows which are prefixed with `pre`. func (db *DB) PGet(pre []byte) (kvs []*KV, err error) { tx, err := db.Txn(false) if err != nil { return } defer tx.Close() return tx.PGet(pre) } // RGet retrieves the range of `max` rows between `beg` (inclusive) and // `end` (exclusive). To return the range in descending order, ensure // that `end` sorts lower than `beg` in the key value store. func (db *DB) RGet(beg, end []byte, max uint64) (kvs []*KV, err error) { tx, err := db.Txn(false) if err != nil { return } defer tx.Close() return tx.RGet(beg, end, max) } // Put sets the value for a key. func (db *DB) Put(key, val []byte) (err error) { tx, err := db.Txn(true) if err != nil { return } defer tx.Commit() return tx.Put(key, val) } // CPut conditionally sets the value for a key if the existing value is equal // to the expected value. To conditionally set a value only if there is no // existing entry pass nil for the expected value. func (db *DB) CPut(key, val, exp []byte) (err error) { tx, err := db.Txn(true) if err != nil { return } defer tx.Commit() return tx.CPut(key, val, exp) } // Del deletes a single key:value item. func (db *DB) Del(key []byte) (err error) { tx, err := db.Txn(true) if err != nil { return } defer tx.Commit() return tx.Del(key) } // CDel conditionally deletes a key if the existing value is equal to the // expected value. func (db *DB) CDel(key, exp []byte) (err error) { tx, err := db.Txn(true) if err != nil { return } defer tx.Commit() return tx.CDel(key, exp) } // MDel deletes multiple key:value items. func (db *DB) MDel(keys ...[]byte) (err error) { tx, err := db.Txn(true) if err != nil { return } defer tx.Commit() return tx.MDel(keys...) } // PDel deletes the range of rows which are prefixed with `pre`. func (db *DB) PDel(pre []byte) (err error) { tx, err := db.Txn(true) if err != nil { return } defer tx.Commit() return tx.PDel(pre) } // RDel deletes the range of `max` rows between `beg` (inclusive) and // `end` (exclusive). To delete the range in descending order, ensure // that `end` sorts lower than `beg` in the key value store. func (db *DB) RDel(beg, end []byte, max uint64) (err error) { tx, err := db.Txn(true) if err != nil { return } defer tx.Commit() return tx.RDel(beg, end, max) } // Txn executes retryable in the context of a distributed transaction. // The transaction is automatically aborted if retryable returns any // error aside from recoverable internal errors, and is automatically // committed otherwise. The retryable function should have no side // effects which could cause problems in the event it must be run more // than once. func (db *DB) Txn(writable bool) (txn *TX, err error) { tx, err := db.db.Begin(writable) if err != nil { tx.Rollback() return } txn = &TX{db: db, tx: tx, bu: tx.Bucket(bucket)} return } func (db *DB) Save(path string) (err error) { tx, err := db.Txn(false) if err != nil { return } defer tx.Close() return tx.tx.CopyFile(path, 0666) } // Close ... func (db *DB) Close() (err error) { // TODO Check if there are transactions open... return db.db.Close() }