store: preparing for full cas support

.gitignore

@@ -0,0 +1 @@
+/example/data

Makefile

@@ -5,4 +5,7 @@ build:
 test:
 	go test -v './...'
 
-.PHONY: build test
+clean:
+	rm -r $(GOPATH)/pkg
+
+.PHONY: build test clean

example/fixtures/articles.go

@@ -6,7 +6,7 @@ import (
 	"time"
 )
 
-const C = 50
+const C = 100000
 
 func main() {
 

example/server/main.go

@@ -4,6 +4,7 @@ import (
 	"simplex.sh/container"
 	_ "simplex.sh/example"
 	_ "simplex.sh/store/file"
+	_ "simplex.sh/store/redis"
 )
 
 func main() { container.CLI() }

shttp/route_handler.go

@@ -47,7 +47,7 @@ func (m *RouteHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 	}
 
 	if rule.Address != "" {
-		body, err = m.store.Get("blobs/" + rule.Address)
+		body, err = m.store.GetBlob("blobs/" + rule.Address)
 		if err != nil {
 			http.Error(w, err.Error(), 500)
 			return
@@ -75,7 +75,7 @@ func (m *RouteHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 
 func (m *RouteHandler) load_routing_table() error {
 
-	r, err := m.store.Get("route_table.json")
+	r, err := m.store.GetBlob("route_table.json")
 	if err != nil {
 		if store.IsNotFound(err) {
 			m.table = map[string][]route_rule{}

shttp/terminator.go

@@ -58,7 +58,7 @@ func (r *terminator) Commit() error {
 func (t *terminator) write_route_table() {
 	defer t.wg.Done()
 
-	w, err := t.tx.DstStore().Set("route_table.json")
+	w, err := t.tx.DstStore().SetBlob("route_table.json")
 	if err != nil {
 		t.err.Add(err)
 		return
@@ -106,7 +106,7 @@ func (t *terminator) document_writer(c <-chan *document) {
 	store := t.tx.DstStore()
 
 	for doc := range c {
-		w, err := store.Set("blobs/" + doc.Digest)
+		w, err := store.SetBlob("blobs/" + doc.Digest)
 		if err != nil {
 			t.err.Add(err)
 			continue

static/tx_coll.go

@@ -34,7 +34,7 @@ func (tx *Tx) Coll(name string, typ interface{}) *C {
 			et = et.Elem()
 		}
 
-		r, err := tx.src.Get(name + ".json")
+		r, err := tx.src.GetBlob(name + ".json")
 		if err != nil {
 			return err
 		}

store/cache.go

@@ -18,11 +18,11 @@ func Cache(sub Store) Store {
 	}
 }
 
-func (c *cache_store) Set(name string) (io.WriteCloser, error) {
-	return c.store.Set(name)
+func (c *cache_store) SetBlob(name string) (io.WriteCloser, error) {
+	return c.store.SetBlob(name)
 }
 
-func (c *cache_store) Get(name string) (io.ReadCloser, error) {
+func (c *cache_store) GetBlob(name string) (io.ReadCloser, error) {
 	if data, p := c.entries[name]; p {
 		return ioutil.NopCloser(bytes.NewReader(data)), nil
 	}
@@ -33,7 +33,7 @@ func (c *cache_store) Get(name string) (io.ReadCloser, error) {
 		err      error
 	)
 
-	upstream, err = c.store.Get(name)
+	upstream, err = c.store.GetBlob(name)
 	if err != nil {
 		return upstream, err
 	}

store/file/file.go → store/file/store.go

@@ -28,7 +28,7 @@ func Open(u *url.URL) (store.Store, error) {
 
 type file_store_t string
 
-func (f file_store_t) Get(name string) (io.ReadCloser, error) {
+func (f file_store_t) GetBlob(name string) (io.ReadCloser, error) {
 	r, err := os.Open(path.Join(string(f), name))
 
 	if err != nil {
@@ -42,7 +42,7 @@ func (f file_store_t) Get(name string) (io.ReadCloser, error) {
 	return r, err
 }
 
-func (f file_store_t) Set(name string) (io.WriteCloser, error) {
+func (f file_store_t) SetBlob(name string) (io.WriteCloser, error) {
 	p := path.Join(string(f), name)
 
 	err := os.MkdirAll(path.Dir(p), 0755)

store/redis/store.go

@@ -0,0 +1,70 @@
+package redis
+
+import (
+	"bytes"
+	"github.com/simonz05/godis/redis"
+	"io"
+	"io/ioutil"
+	"net/url"
+	"path"
+	"simplex.sh/store"
+)
+
+func init() {
+	store.Register("redis", Open)
+}
+
+type store_t struct {
+	conn   *redis.Client
+	prefix string
+}
+
+type setter struct {
+	io.Writer
+	conn *redis.Client
+	key  string
+}
+
+func Open(u *url.URL) (store.Store, error) {
+	var (
+		addr   = "tcp:" + u.Host
+		pass   = ""
+		prefix = "sx:"
+		conn   *redis.Client
+	)
+
+	if ui := u.User; ui != nil {
+		pass, _ = ui.Password()
+	}
+
+	prefix += u.Path
+	prefix = path.Clean(prefix)
+
+	conn = redis.New(addr, 0, pass)
+
+	return &store_t{conn, prefix}, nil
+}
+
+func (s *store_t) GetBlob(name string) (io.ReadCloser, error) {
+	key := path.Join(s.prefix, name)
+	elem, err := s.conn.Get(key)
+
+	if err != nil {
+		return nil, err
+	}
+
+	if elem == nil {
+		return nil, store.NotFoundError(name)
+	}
+
+	return ioutil.NopCloser(bytes.NewReader(elem.Bytes())), err
+}
+
+func (s *store_t) SetBlob(name string) (io.WriteCloser, error) {
+	key := path.Join(s.prefix, name)
+	return &setter{bytes.NewBuffer(nil), s.conn, key}, nil
+}
+
+func (s *setter) Close() error {
+	return s.conn.Set(s.key, s.Writer.(*bytes.Buffer).Bytes())
+}

store/store.go

@@ -5,8 +5,8 @@ import (
 )
 
 type Store interface {
-	Get(name string) (io.ReadCloser, error)
-	Set(name string) (io.WriteCloser, error)
+	GetBlob(name string) (io.ReadCloser, error)
+	SetBlob(name string) (io.WriteCloser, error)
 }
 
 type notfound_error struct {

store/sub.go

@@ -14,10 +14,10 @@ func SubStore(store Store, prefix string) Store {
 	return &sub_store_t{prefix, store}
 }
 
-func (s *sub_store_t) Get(name string) (io.ReadCloser, error) {
-	return s.store.Get(path.Join(s.prefix, name))
+func (s *sub_store_t) GetBlob(name string) (io.ReadCloser, error) {
+	return s.store.GetBlob(path.Join(s.prefix, name))
 }
 
-func (s *sub_store_t) Set(name string) (io.WriteCloser, error) {
-	return s.store.Set(path.Join(s.prefix, name))
+func (s *sub_store_t) SetBlob(name string) (io.WriteCloser, error) {
+	return s.store.SetBlob(path.Join(s.prefix, name))
 }