add log abstraction to store records in disk

Author: Hongchao Deng

db/backend.go

@@ -1,45 +1,35 @@
 package db
 
 import (
-	"io/ioutil"
-	"os"
 	"strings"
 
-	"github.com/go-distributed/xtree/db/recordio"
+	"github.com/go-distributed/xtree/db/log"
+	"github.com/go-distributed/xtree/db/message"
 	"github.com/go-distributed/xtree/third-party/github.com/google/btree"
 )
 
 type backend struct {
 	bt    *btree.BTree
 	cache *cache
 	rev   int
-	fc    recordio.Fetcher
-	ap    recordio.Appender
+	log   *log.Log
 }
 
 func newBackend() *backend {
 	bt := btree.New(10)
-
-	// temporary file IO to test in-disk values
-	writeFile, err := ioutil.TempFile("", "backend")
-	if err != nil {
-		panic("can't create temp file")
-	}
-	readFile, err := os.Open(writeFile.Name())
+	log, err := log.Create()
 	if err != nil {
-		panic("can't open temp file")
+		panic("Not implemented")
 	}
-
 	return &backend{
 		bt:    bt,
 		cache: newCache(),
-		fc:    recordio.NewFetcher(readFile),
-		ap:    recordio.NewAppender(writeFile),
+		log:   log,
 	}
 }
 
 func (b *backend) getData(offset int64) []byte {
-	rec, err := b.fc.Fetch(offset)
+	rec, err := b.log.GetRecord(offset)
 	if err != nil {
 		panic("unimplemented")
 	}
@@ -87,7 +77,11 @@ func (b *backend) Put(rev int, path Path, data []byte) {
 	}
 
 	b.rev++
-	offset, err := b.ap.Append(recordio.Record{data})
+	offset, err := b.log.Append(&message.Record{
+		Rev:  b.rev,
+		Key:  path.p,
+		Data: data,
+	})
 	if err != nil {
 		panic("unimplemented")
 	}

db/db.go

@@ -23,7 +23,7 @@ type DB interface {
 	// Otherwise, it lists recursively all paths.
 	//
 	// if count is >= 0, it is the number of paths we want in the list.
-	// if count is -1, it means any.
+	// if count is -1, it means all.
 	//
 	// if it failed, an error is returned.
 	Ls(rev int, path string, recursive bool, count int) ([]Path, error)

db/log/decoder.go

@@ -0,0 +1,34 @@
+package log
+
+import (
+	"bufio"
+	"encoding/binary"
+	"io"
+
+	"github.com/go-distributed/xtree/db/message"
+)
+
+type decoder struct {
+	br *bufio.Reader
+}
+
+func newDecoder(r io.Reader) *decoder {
+	return &decoder{bufio.NewReader(r)}
+}
+
+func (d *decoder) decode(r *message.Record) (err error) {
+	var l int64
+	if l, err = readInt64(d.br); err != nil {
+		return
+	}
+	data := make([]byte, l)
+	if _, err = io.ReadFull(d.br, data); err != nil {
+		return
+	}
+	return r.Unmarshal(data)
+}
+
+func readInt64(r io.Reader) (n int64, err error) {
+	err = binary.Read(r, binary.LittleEndian, &n)
+	return
+}

db/log/encoder.go

@@ -0,0 +1,37 @@
+package log
+
+import (
+	"bufio"
+	"encoding/binary"
+	"io"
+
+	"github.com/go-distributed/xtree/db/message"
+)
+
+type encoder struct {
+	bw *bufio.Writer
+}
+
+func newEncoder(w io.Writer) *encoder {
+	return &encoder{bufio.NewWriter(w)}
+}
+
+func (e *encoder) encode(r *message.Record) (err error) {
+	var data []byte
+	if data, err = r.Marshal(); err != nil {
+		return
+	}
+	if err = writeInt64(e.bw, int64(len(data))); err != nil {
+		return
+	}
+	_, err = e.bw.Write(data)
+	return
+}
+
+func (e *encoder) flush() error {
+	return e.bw.Flush()
+}
+
+func writeInt64(w io.Writer, n int64) error {
+	return binary.Write(w, binary.LittleEndian, n)
+}

db/log/encoder_test.go

@@ -0,0 +1,47 @@
+package log
+
+import (
+	"bytes"
+	"reflect"
+	"testing"
+
+	"github.com/go-distributed/xtree/db/message"
+)
+
+func TestEncoderDecoder(t *testing.T) {
+	tests := []struct {
+		rec *message.Record
+	}{
+		{&message.Record{
+			Rev:  1,
+			Key:  "/test",
+			Data: []byte("some data"),
+		}},
+	}
+
+	for i, tt := range tests {
+		var err error
+		eBuf := new(bytes.Buffer)
+		encoder := newEncoder(eBuf)
+
+		if err = encoder.encode(tt.rec); err != nil {
+			t.Fatalf("#%d: cannot encode, err: %v", i, err)
+		}
+		if err = encoder.flush(); err != nil {
+			t.Fatalf("#%d: cannot flush encode, err: %v", i, err)
+		}
+
+		rec := &message.Record{}
+		dBuf := bytes.NewBuffer(eBuf.Bytes())
+		decoder := newDecoder(dBuf)
+
+		if err = decoder.decode(rec); err != nil {
+			t.Fatalf("#%d: cannot decode, err: %v", i, err)
+		}
+
+		if !reflect.DeepEqual(tt.rec, rec) {
+			t.Fatalf("#%d: records are not the same, want: %v, get: %v",
+				i, tt.rec, rec)
+		}
+	}
+}

db/log/log.go

@@ -0,0 +1,51 @@
+package log
+
+import (
+	"io/ioutil"
+	"os"
+
+	"github.com/go-distributed/xtree/db/message"
+)
+
+type Log struct {
+	f *os.File
+}
+
+func Create() (*Log, error) {
+	f, err := ioutil.TempFile("", "backend")
+	if err != nil {
+		return nil, err
+	}
+
+	return &Log{f}, nil
+}
+
+func (l *Log) Destroy() error {
+	return os.Remove(l.f.Name())
+}
+
+func (l *Log) GetRecord(offset int64) (*message.Record, error) {
+	_, err := l.f.Seek(offset, 0)
+	if err != nil {
+		return nil, err
+	}
+
+	decoder := newDecoder(l.f)
+	r := &message.Record{}
+	decoder.decode(r)
+
+	return r, nil
+}
+
+func (l *Log) Append(r *message.Record) (offset int64, err error) {
+	offset, err = l.f.Seek(0, 2)
+	if err != nil {
+		return -1, err
+	}
+
+	encoder := newEncoder(l.f)
+
+	err = encoder.encode(r)
+	encoder.flush()
+	return offset, err
+}

db/log/log_test.go

@@ -0,0 +1,54 @@
+package log
+
+import (
+	"reflect"
+	"testing"
+
+	"github.com/go-distributed/xtree/db/message"
+)
+
+func TestAppendAndGetRecord(t *testing.T) {
+	var err error
+	var log *Log
+	if log, err = Create(); err != nil {
+		t.Errorf("Create failed: %v", err)
+	}
+
+	defer log.Destroy()
+
+	tests := []struct {
+		offset int64
+		rec    *message.Record
+	}{
+		{-1, &message.Record{
+			Rev:  1,
+			Key:  "/test",
+			Data: []byte("some data"),
+		}},
+		{-1, &message.Record{
+			Rev:  2,
+			Key:  "/test2",
+			Data: []byte("some other data"),
+		}},
+	}
+
+	for i, tt := range tests {
+		tests[i].offset, err = log.Append(tt.rec)
+		if err != nil {
+			t.Errorf("#%d: Append failed: %v", i, err)
+		}
+	}
+
+	for i, tt := range tests {
+		var rec *message.Record
+		if rec, err = log.GetRecord(tt.offset); err != nil {
+			t.Errorf("#%d: GetRecord failed: %v", i, err)
+		}
+
+		if !reflect.DeepEqual(tt.rec, rec) {
+			t.Errorf("#%d: records not the same, want: %v, get %v",
+				i, tt.rec, rec)
+		}
+
+	}
+}

db/message/record.go

@@ -0,0 +1,17 @@
+package message
+
+import "encoding/json"
+
+type Record struct {
+	Rev  int
+	Key  string
+	Data []byte
+}
+
+func (r *Record) Marshal() ([]byte, error) {
+	return json.Marshal(r)
+}
+
+func (r *Record) Unmarshal(data []byte) error {
+	return json.Unmarshal(data, r)
+}

db/message/record_test.go

@@ -0,0 +1,36 @@
+package message
+
+import (
+	"reflect"
+	"testing"
+)
+
+func TestLogRecord(t *testing.T) {
+	tests := []struct {
+		rec *Record
+	}{
+		{&Record{
+			Rev:  1,
+			Key:  "/test",
+			Data: []byte("some data"),
+		}},
+	}
+
+	for i, tt := range tests {
+		var data []byte
+		var err error
+		if data, err = tt.rec.Marshal(); err != nil {
+			t.Fatalf("#%d: cannot marshal, err: %v", i, err)
+		}
+
+		rec := &Record{}
+		if err = rec.Unmarshal(data); err != nil {
+			t.Fatalf("#%d: cannot unmarshal, err: %v", i, err)
+		}
+
+		if !reflect.DeepEqual(tt.rec, rec) {
+			t.Fatalf("#%d: records are not the same, want: %v, get: %v",
+				i, tt.rec, rec)
+		}
+	}
+}