Skip to content

GitLab

Commit a4a424f7 authored by Jeromy's avatar Jeromy
Browse files
Options

extract leveldb datastore

parent 3b6efa41
Hide whitespace changes
Inline Side-by-side
Showing with 3 additions and 354 deletions
leveldb/datastore.go deleted 100644 → 0
View file @ 3b6efa41
package leveldb
import (
ds "github.com/ipfs/go-datastore"
dsq "github.com/ipfs/go-datastore/query"
"github.com/jbenet/goprocess"
"github.com/syndtr/goleveldb/leveldb"
"github.com/syndtr/goleveldb/leveldb/opt"
"github.com/syndtr/goleveldb/leveldb/util"
)
type datastore struct {
DB *leveldb.DB
}
type Options opt.Options
func NewDatastore(path string, opts *Options) (*datastore, error) {
var nopts opt.Options
if opts != nil {
nopts = opt.Options(*opts)
}
db, err := leveldb.OpenFile(path, &nopts)
if err != nil {
return nil, err
}
return &datastore{
DB: db,
}, nil
}
// Returns ErrInvalidType if value is not of type []byte.
//
// Note: using sync = false.
// see http://godoc.org/github.com/syndtr/goleveldb/leveldb/opt#WriteOptions
func (d *datastore) Put(key ds.Key, value interface{}) (err error) {
val, ok := value.([]byte)
if !ok {
return ds.ErrInvalidType
}
return d.DB.Put(key.Bytes(), val, nil)
}
func (d *datastore) Get(key ds.Key) (value interface{}, err error) {
val, err := d.DB.Get(key.Bytes(), nil)
if err != nil {
if err == leveldb.ErrNotFound {
return nil, ds.ErrNotFound
}
return nil, err
}
return val, nil
}
func (d *datastore) Has(key ds.Key) (exists bool, err error) {
return d.DB.Has(key.Bytes(), nil)
}
func (d *datastore) Delete(key ds.Key) (err error) {
// leveldb Delete will not return an error if the key doesn't
// exist (see https://github.com/syndtr/goleveldb/issues/109),
// so check that the key exists first and if not return an
// error
exists, err := d.DB.Has(key.Bytes(), nil)
if !exists {
return ds.ErrNotFound
} else if err != nil {
return err
}
return d.DB.Delete(key.Bytes(), nil)
}
func (d *datastore) Query(q dsq.Query) (dsq.Results, error) {
// we can use multiple iterators concurrently. see:
// https://godoc.org/github.com/syndtr/goleveldb/leveldb#DB.NewIterator
// advance the iterator only if the reader reads
//
// run query in own sub-process tied to Results.Process(), so that
// it waits for us to finish AND so that clients can signal to us
// that resources should be reclaimed.
qrb := dsq.NewResultBuilder(q)
qrb.Process.Go(func(worker goprocess.Process) {
d.runQuery(worker, qrb)
})
// go wait on the worker (without signaling close)
go qrb.Process.CloseAfterChildren()
// Now, apply remaining things (filters, order)
qr := qrb.Results()
for _, f := range q.Filters {
qr = dsq.NaiveFilter(qr, f)
}
for _, o := range q.Orders {
qr = dsq.NaiveOrder(qr, o)
}
return qr, nil
}
func (d *datastore) runQuery(worker goprocess.Process, qrb *dsq.ResultBuilder) {
var rnge *util.Range
if qrb.Query.Prefix != "" {
rnge = util.BytesPrefix([]byte(qrb.Query.Prefix))
}
i := d.DB.NewIterator(rnge, nil)
defer i.Release()
// advance iterator for offset
if qrb.Query.Offset > 0 {
for j := 0; j < qrb.Query.Offset; j++ {
i.Next()
}
}
// iterate, and handle limit, too
for sent := 0; i.Next(); sent++ {
// end early if we hit the limit
if qrb.Query.Limit > 0 && sent >= qrb.Query.Limit {
break
}
k := ds.NewKey(string(i.Key())).String()
e := dsq.Entry{Key: k}
if !qrb.Query.KeysOnly {
buf := make([]byte, len(i.Value()))
copy(buf, i.Value())
e.Value = buf
}
select {
case qrb.Output <- dsq.Result{Entry: e}: // we sent it out
case <-worker.Closing(): // client told us to end early.
break
}
}
if err := i.Error(); err != nil {
select {
case qrb.Output <- dsq.Result{Error: err}: // client read our error
case <-worker.Closing(): // client told us to end.
return
}
}
}
// LevelDB needs to be closed.
func (d *datastore) Close() (err error) {
return d.DB.Close()
}
func (d *datastore) IsThreadSafe() {}
type leveldbBatch struct {
b *leveldb.Batch
db *leveldb.DB
}
func (d *datastore) Batch() (ds.Batch, error) {
return &leveldbBatch{
b: new(leveldb.Batch),
db: d.DB,
}, nil
}
func (b *leveldbBatch) Put(key ds.Key, value interface{}) error {
val, ok := value.([]byte)
if !ok {
return ds.ErrInvalidType
}
b.b.Put(key.Bytes(), val)
return nil
}
func (b *leveldbBatch) Commit() error {
return b.db.Write(b.b, nil)
}
func (b *leveldbBatch) Delete(key ds.Key) error {
b.b.Delete(key.Bytes())
return nil
}
leveldb/ds_test.go deleted 100644 → 0
View file @ 3b6efa41
package leveldb
import (
"io/ioutil"
"os"
"testing"
ds "github.com/ipfs/go-datastore"
dsq "github.com/ipfs/go-datastore/query"
)
var testcases = map[string]string{
"/a": "a",
"/a/b": "ab",
"/a/b/c": "abc",
"/a/b/d": "a/b/d",
"/a/c": "ac",
"/a/d": "ad",
"/e": "e",
"/f": "f",
}
// returns datastore, and a function to call on exit.
// (this garbage collects). So:
//
// d, close := newDS(t)
// defer close()
func newDS(t *testing.T) (*datastore, func()) {
path, err := ioutil.TempDir("/tmp", "testing_leveldb_")
if err != nil {
t.Fatal(err)
}
d, err := NewDatastore(path, nil)
if err != nil {
t.Fatal(err)
}
return d, func() {
os.RemoveAll(path)
d.Close()
}
}
func addTestCases(t *testing.T, d *datastore, testcases map[string]string) {
for k, v := range testcases {
dsk := ds.NewKey(k)
if err := d.Put(dsk, []byte(v)); err != nil {
t.Fatal(err)
}
}
for k, v := range testcases {
dsk := ds.NewKey(k)
v2, err := d.Get(dsk)
if err != nil {
t.Fatal(err)
}
v2b := v2.([]byte)
if string(v2b) != v {
t.Errorf("%s values differ: %s != %s", k, v, v2)
}
}
}
func TestQuery(t *testing.T) {
d, close := newDS(t)
defer close()
addTestCases(t, d, testcases)
rs, err := d.Query(dsq.Query{Prefix: "/a/"})
if err != nil {
t.Fatal(err)
}
expectMatches(t, []string{
"/a/b",
"/a/b/c",
"/a/b/d",
"/a/c",
"/a/d",
}, rs)
// test offset and limit
rs, err = d.Query(dsq.Query{Prefix: "/a/", Offset: 2, Limit: 2})
if err != nil {
t.Fatal(err)
}
expectMatches(t, []string{
"/a/b/d",
"/a/c",
}, rs)
}
func TestQueryRespectsProcess(t *testing.T) {
d, close := newDS(t)
defer close()
addTestCases(t, d, testcases)
}
func expectMatches(t *testing.T, expect []string, actualR dsq.Results) {
actual, err := actualR.Rest()
if err != nil {
t.Error(err)
}
if len(actual) != len(expect) {
t.Error("not enough", expect, actual)
}
for _, k := range expect {
found := false
for _, e := range actual {
if e.Key == k {
found = true
}
}
if !found {
t.Error(k, "not found")
}
}
}
func TestBatching(t *testing.T) {
d, done := newDS(t)
defer done()
b, err := d.Batch()
if err != nil {
t.Fatal(err)
}
for k, v := range testcases {
err := b.Put(ds.NewKey(k), []byte(v))
if err != nil {
t.Fatal(err)
}
}
err = b.Commit()
if err != nil {
t.Fatal(err)
}
for k, v := range testcases {
val, err := d.Get(ds.NewKey(k))
if err != nil {
t.Fatal(err)
}
if v != string(val.([]byte)) {
t.Fatal("got wrong data!")
}
}
}
package.json
View file @ a4a424f7
......@@ -13,12 +13,6 @@
"name": "go.uuid",
"version": "1.0.0"
},
{
"author": "syndtr",
"hash": "QmbBhyDKsY4mbY6xsKt3qu9Y7FPvMJ6qbD8AMjYYvPRw1g",
"name": "goleveldb",
"version": "0.0.1"
},
{
"author": "whyrusleeping",
"hash": "QmQopLATEYMNg7dVqZRNDfeE2S1yKy8zrRh5xnYiuqeZBn",
......
test/test_util.go
View file @ a4a424f7
......@@ -3,9 +3,9 @@ package dstest
import (
"bytes"
"encoding/base32"
"math/rand"
"testing"
rand "github.com/dustin/randbo"
dstore "github.com/ipfs/go-datastore"
)
......@@ -15,12 +15,11 @@ func RunBatchTest(t *testing.T, ds dstore.Batching) {
t.Fatal(err)
}
r := rand.New()
var blocks [][]byte
var keys []dstore.Key
for i := 0; i < 20; i++ {
blk := make([]byte, 256*1024)
r.Read(blk)
rand.Read(blk)
blocks = append(blocks, blk)
key := dstore.NewKey(base32.StdEncoding.EncodeToString(blk[:8]))
......@@ -59,11 +58,10 @@ func RunBatchTest(t *testing.T, ds dstore.Batching) {
}
func RunBatchDeleteTest(t *testing.T, ds dstore.Batching) {
r := rand.New()
var keys []dstore.Key
for i := 0; i < 20; i++ {
blk := make([]byte, 16)
r.Read(blk)
rand.Read(blk)
key := dstore.NewKey(base32.StdEncoding.EncodeToString(blk[:8]))
keys = append(keys, key)
......
Markdown is supported
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment