Added transcoder functions to LevelDB backend.

This commit is contained in:
Sascha L. Teichmann 2014-08-22 22:26:03 +02:00
parent 24b9236e41
commit 712708f87c
3 changed files with 39 additions and 15 deletions

View File

@ -24,6 +24,7 @@ type (
KeyTransformer func(int64) int64 KeyTransformer func(int64) int64
KeyEncoder func(int64) ([]byte, error) KeyEncoder func(int64) ([]byte, error)
KeyDecoder func([]byte) (int64, error) KeyDecoder func([]byte) (int64, error)
KeyTranscoder func([]byte) ([]byte, error)
KeySplitter func(int64) Coord KeySplitter func(int64) Coord
KeyJoiner func(Coord) int64 KeyJoiner func(Coord) int64
) )
@ -156,6 +157,10 @@ func EncodeStringToBytesFromInterleaved(key int64) ([]byte, error) {
return EncodeStringToBytes(TransformInterleavedToPlain(key)) return EncodeStringToBytes(TransformInterleavedToPlain(key))
} }
func IdentityTranscoder(key []byte) ([]byte, error) {
return key, nil
}
func SelectKeySplitter(interleaved bool) KeySplitter { func SelectKeySplitter(interleaved bool) KeySplitter {
if interleaved { if interleaved {
return InterleavedToCoord return InterleavedToCoord

View File

@ -8,13 +8,17 @@ import (
"log" "log"
"sync" "sync"
"bitbucket.org/s_l_teichmann/mtredisalize/common"
leveldb "github.com/jmhodges/levigo" leveldb "github.com/jmhodges/levigo"
) )
type LevelDBBackend struct { type LevelDBBackend struct {
cache *leveldb.Cache cache *leveldb.Cache
db *leveldb.DB db *leveldb.DB
mutex sync.RWMutex encoder common.KeyTranscoder
decoder common.KeyTranscoder
mutex sync.RWMutex
} }
type LevelDBSession struct { type LevelDBSession struct {
@ -22,7 +26,11 @@ type LevelDBSession struct {
tx *leveldb.WriteBatch tx *leveldb.WriteBatch
} }
func NewLeveDBBackend(path string, cacheSize int) (ldb *LevelDBBackend, err error) { func NewLeveDBBackend(
path string,
encoder common.KeyTranscoder,
decoder common.KeyTranscoder,
cacheSize int) (ldb *LevelDBBackend, err error) {
opts := leveldb.NewOptions() opts := leveldb.NewOptions()
var cache *leveldb.Cache var cache *leveldb.Cache
@ -41,8 +49,10 @@ func NewLeveDBBackend(path string, cacheSize int) (ldb *LevelDBBackend, err erro
return return
} }
ldb = &LevelDBBackend{ ldb = &LevelDBBackend{
cache: cache, cache: cache,
db: db} db: db,
encoder: encoder,
decoder: decoder}
return return
} }
@ -78,7 +88,9 @@ func (ldb *LevelDBBackend) Shutdown() error {
} }
func (ldbs *LevelDBSession) Fetch(hash, key []byte) (value []byte, err error) { func (ldbs *LevelDBSession) Fetch(hash, key []byte) (value []byte, err error) {
if key, err = ldbs.backend.decoder(key); err != nil {
return
}
ldbs.backend.doRead(func(db *leveldb.DB) { ldbs.backend.doRead(func(db *leveldb.DB) {
ro := leveldb.NewReadOptions() ro := leveldb.NewReadOptions()
value, err = ldbs.backend.db.Get(ro, key) value, err = ldbs.backend.db.Get(ro, key)
@ -108,24 +120,21 @@ func keyExists(db *leveldb.DB, key []byte) (exists bool, err error) {
} }
func (ldbs *LevelDBSession) Store(hash, key, value []byte) (exists bool, err error) { func (ldbs *LevelDBSession) Store(hash, key, value []byte) (exists bool, err error) {
if key, err = ldbs.backend.decoder(key); err != nil {
return
}
ldbs.backend.doWrite(func(db *leveldb.DB) { ldbs.backend.doWrite(func(db *leveldb.DB) {
if exists, err = keyExists(db, key); err != nil { if exists, err = keyExists(db, key); err != nil {
return return
} }
if ldbs.tx != nil { if ldbs.tx != nil {
//log.Printf("Stored in tx: key = '%s' len(value) = %d\n", key, len(value))
ldbs.tx.Put(key, value) ldbs.tx.Put(key, value)
return return
} }
//log.Printf("Stored: key = '%s' len(value) = %d\n", key, len(value))
wo := leveldb.NewWriteOptions() wo := leveldb.NewWriteOptions()
err = ldbs.backend.db.Put(wo, key, value) err = ldbs.backend.db.Put(wo, key, value)
wo.Close() wo.Close()
}) })
return return
} }
@ -178,8 +187,14 @@ func (ldbs *LevelDBSession) AllKeys(hash []byte) (keys chan []byte, n int, err e
defer close(keys) defer close(keys)
defer it.Close() defer it.Close()
it.SeekToFirst() it.SeekToFirst()
encoder := ldbs.backend.encoder
for ; it.Valid(); it.Next() { for ; it.Valid(); it.Next() {
keys <- it.Key() if key, err := encoder(it.Key()); err == nil {
keys <- key
} else {
log.Printf("WARN: %s\n", err)
return
}
} }
if err := it.GetError(); err != nil { if err := it.GetError(); err != nil {
log.Printf("WARN: %s\n", err) log.Printf("WARN: %s\n", err)

View File

@ -70,7 +70,11 @@ func main() {
log.Fatal(err) log.Fatal(err)
} }
} else { } else {
if backend, err = NewLeveDBBackend(args[0], cacheSize); err != nil { if backend, err = NewLeveDBBackend(
args[0],
common.IdentityTranscoder,
common.IdentityTranscoder,
cacheSize); err != nil {
log.Fatal(err) log.Fatal(err)
} }
} }