Create dbstream interface

files.go

@@ -1,6 +1,8 @@
 package storage
 
 import (
+	"bytes"
+	"io"
 	"io/ioutil"
 	"os"
 	"path"
@@ -49,15 +51,31 @@ func (b *Files) List(ns []string, limits ...string) ([]string, error) {
 }
 
 func (b *Files) Get(key string, ns ...string) ([]byte, error) {
+	r, err := b.GetStream(key, ns...)
+	if err != nil {
+		return nil, err
+	}
+	return ioutil.ReadAll(r)
+}
+
+func (b *Files) GetStream(key string, ns ...string) (io.Reader, error) {
 	namespace := resolveNamespace(ns)
 	path := path.Join(b.root, namespace, key)
-	return ioutil.ReadFile(path)
+	return os.Open(path)
 }
 
 func (b *Files) Set(key string, value []byte, ns ...string) error {
+	r := bytes.NewReader(value)
+	if value == nil {
+		r = nil
+	}
+	return b.SetStream(key, r, ns...)
+}
+
+func (b *Files) SetStream(key string, r io.Reader, ns ...string) error {
 	namespace := resolveNamespace(ns)
 	dir := path.Join(b.root, namespace)
-	if value == nil {
+	if r == nil {
 		err := os.Remove(path.Join(dir, key))
 		if os.IsNotExist(err) {
 			err = nil
@@ -68,7 +86,13 @@ func (b *Files) Set(key string, value []byte, ns ...string) error {
 		return err
 	}
 	path := path.Join(dir, key)
-	return ioutil.WriteFile(path, value, os.ModePerm)
+	f, err := os.Create(path)
+	if err != nil {
+		return err
+	}
+	defer f.Close()
+	_, err = io.Copy(f, r)
+	return err
 }
 
 func (b *Files) Close() error {