feat: database backend add ssdb support

Change-Id: I054c5fc9b02f613601781de8613d684faa0ea7f2

Author: hudeng-go

AUTHORS

@@ -52,3 +52,4 @@ List of contributors, in chronological order:
 * Steven Stone (https://github.com/smstone)
 * Josh Bayfield (https://github.com/jbayfield)
 * Boxjan (https://github.com/boxjan)
+* goldendeng (https://github.com/goldendeng)

context/context.go

@@ -3,13 +3,16 @@ package context
 
 import (
 	gocontext "context"
+	"errors"
 	"fmt"
 	"math/rand"
+	"net/url"
 	"os"
 	"os/signal"
 	"path/filepath"
 	"runtime"
 	"runtime/pprof"
+	"strconv"
 	"strings"
 	"sync"
 	"time"
@@ -19,6 +22,7 @@ import (
 	"github.com/aptly-dev/aptly/console"
 	"github.com/aptly-dev/aptly/database"
 	"github.com/aptly-dev/aptly/database/goleveldb"
+	"github.com/aptly-dev/aptly/database/ssdb"
 	"github.com/aptly-dev/aptly/deb"
 	"github.com/aptly-dev/aptly/files"
 	"github.com/aptly-dev/aptly/http"
@@ -27,6 +31,7 @@ import (
 	"github.com/aptly-dev/aptly/swift"
 	"github.com/aptly-dev/aptly/task"
 	"github.com/aptly-dev/aptly/utils"
+	"github.com/seefan/gossdb/v2/conf"
 	"github.com/smira/commander"
 	"github.com/smira/flag"
 )
@@ -287,7 +292,32 @@ func (context *AptlyContext) _database() (database.Storage, error) {
 	if context.database == nil {
 		var err error
 
-		context.database, err = goleveldb.NewDB(context.dbPath())
+		if context.config().DatabaseBackend.Type == "leveldb" {
+			if context.config().DatabaseBackend.DbPath != "" {
+				dbPath := filepath.Join(context.config().RootDir, context.config().DatabaseBackend.DbPath)
+				context.database, err = goleveldb.NewDB(dbPath)
+			} else {
+				return nil, errors.New("leveldb databaseBackend config invalid")
+			}
+		} else if context.config().DatabaseBackend.Type == "ssdb" {
+			var cfg conf.Config
+			u, e := url.Parse(context.config().DatabaseBackend.URL)
+
+			if e != nil {
+				return nil, e
+			}
+			cfg.Port, e = strconv.Atoi(u.Port())
+			cfg.Host = strings.Split(u.Host, ":")[0]
+			if e != nil {
+				return nil, e
+			}
+			password, _ := u.User.Password()
+			cfg.Password = password
+			context.database, err = ssdb.NewOpenDB(&cfg)
+		} else {
+			context.database, err = goleveldb.NewDB(context.dbPath())
+		}
+
 		if err != nil {
 			return nil, fmt.Errorf("can't instantiate database: %s", err)
 		}

database/ssdb/batch.go

@@ -0,0 +1,125 @@
+package ssdb
+
+import (
+	"fmt"
+
+	"github.com/aptly-dev/aptly/database"
+	"github.com/seefan/gossdb/v2/conf"
+	"github.com/seefan/gossdb/v2/pool"
+)
+
+type bWriteData struct {
+	key   []byte
+	value []byte
+	opts  string
+	err   error
+}
+
+type SsdbBatch struct {
+	cfg *conf.Config
+	// key-value chan
+	w  chan bWriteData
+	p  map[string]interface{}
+	d  []string
+	db *pool.Client
+}
+
+// func internalOpenBatch...
+func internalOpenBatch(t database.Storage) *SsdbBatch {
+	b := &SsdbBatch{
+		w: make(chan bWriteData),
+		p: make(map[string]interface{}),
+	}
+	b.run()
+
+	return b
+}
+
+func (b *SsdbBatch) run() {
+	go func() {
+		for {
+			select {
+			case w, ok := <-b.w:
+				{
+					if !ok {
+						ssdb_log("ssdb batch write chan closed")
+						return
+					}
+
+					if w.opts == "write" {
+						ssdb_log("ssdb batch write")
+						var err error
+						if len(b.p) > 0 && len(b.d) == 0 {
+							err = b.db.MultiSet(b.p)
+							ssdb_log("ssdb batch set errinfo: ", err)
+						} else if len(b.d) > 0 && len(b.p) == 0 {
+							err = b.db.MultiDel(b.d...)
+							ssdb_log("ssdb batch del errinfo: ", err)
+						} else if len(b.p) == 0 && len(b.d) == 0 {
+							err = nil
+						} else {
+							err = fmt.Errorf("ssdb batch does not support both put and delete operations")
+						}
+						ssdb_log("ssdb batch write errinfo: ", err)
+						b.w <- bWriteData{
+							err: err,
+						}
+						ssdb_log("ssdb batch write end")
+					} else {
+						ssdb_log("ssdb batch", w.opts)
+						if w.opts == "put" {
+							b.p[string(w.key)] = w.value
+						} else if w.opts == "del" {
+							b.d = append(b.d, string(w.key))
+						}
+					}
+				}
+			}
+		}
+	}()
+}
+
+func (b *SsdbBatch) stop() {
+	ssdb_log("ssdb batch stop")
+	close(b.w)
+}
+
+func (b *SsdbBatch) Put(key, value []byte) (err error) {
+	// err = b.db.Set(string(key), string(value))
+	w := bWriteData{
+		key:   key,
+		value: value,
+		opts:  "put",
+	}
+
+	b.w <- w
+	return nil
+}
+
+func (b *SsdbBatch) Delete(key []byte) (err error) {
+	/* err = b.db.Del(string(key))
+	return */
+	w := bWriteData{
+		key:  key,
+		opts: "del",
+	}
+
+	b.w <- w
+	return nil
+}
+
+func (b *SsdbBatch) Write() (err error) {
+	defer b.stop()
+	w := bWriteData{
+		opts: "write",
+	}
+
+	b.w <- w
+	result := <-b.w
+	return result.err
+}
+
+// batch should implement database.Batch
+var (
+	_ database.Batch = &SsdbBatch{}
+)

database/ssdb/database.go

@@ -0,0 +1,62 @@
+package ssdb
+
+import (
+	"os"
+	"strconv"
+
+	"github.com/aptly-dev/aptly/database"
+	"github.com/seefan/gossdb/v2"
+	"github.com/seefan/gossdb/v2/conf"
+	"github.com/seefan/gossdb/v2/pool"
+)
+
+var defaultBufSize = 102400
+var defaultPoolSize = 1
+
+func internalOpen(cfg *conf.Config) (*pool.Client, error) {
+	ssdb_log("internalOpen")
+
+	cfg.ReadBufferSize = defaultBufSize
+	cfg.WriteBufferSize = defaultBufSize
+	cfg.MaxPoolSize = defaultPoolSize
+	cfg.PoolSize = defaultPoolSize
+	cfg.MinPoolSize = defaultPoolSize
+	cfg.MaxWaitSize = 100 * defaultPoolSize
+	cfg.RetryEnabled = true
+
+	//override by env
+	if os.Getenv("SSDB_READBUFFERSIZE") != "" {
+		readBufSize, err := strconv.Atoi(os.Getenv("SSDB_READBUFFERSIZE"))
+		if err != nil {
+			cfg.ReadBufferSize = readBufSize
+		}
+	}
+
+	if os.Getenv("SSDB_WRITEBUFFERSIZE") != "" {
+		writeBufSize, err := strconv.Atoi(os.Getenv("SSDB_WRITEBUFFERSIZE"))
+		if err != nil {
+			cfg.WriteBufferSize = writeBufSize
+		}
+	}
+
+	var cfgs = []*conf.Config{cfg}
+	err := gossdb.Start(cfgs...)
+	if err != nil {
+		return nil, err
+	}
+
+	return gossdb.NewClient()
+}
+
+func NewDB(cfg *conf.Config) (database.Storage, error) {
+	return &SsdbStorage{cfg: cfg}, nil
+}
+
+func NewOpenDB(cfg *conf.Config) (database.Storage, error) {
+	db, err := NewDB(cfg)
+	if err != nil {
+		return nil, err
+	}
+
+	return db, db.Open()
+}