From 0a45ec1d3a171d94cc65a6a43e0c7414e5532734 Mon Sep 17 00:00:00 2001
From: Alexander Neumann <alexander@bumpern.de>
Date: Wed, 11 Feb 2015 19:50:29 +0100
Subject: [PATCH] Implement streaming functions for local backend

---
 backend/local.go      | 66 +++++++++++++++++++++++++++++++++++++++++++
 backend/local_test.go | 27 ++++++++++++++++++
 2 files changed, 93 insertions(+)

diff --git a/backend/local.go b/backend/local.go
index 97dfe092f..6b7402287 100644
--- a/backend/local.go
+++ b/backend/local.go
@@ -3,6 +3,7 @@ package backend
 import (
 	"errors"
 	"fmt"
+	"io"
 	"io/ioutil"
 	"os"
 	"path/filepath"
@@ -223,6 +224,54 @@ func (b *Local) Create(t Type, data []byte) (ID, error) {
 	return id, nil
 }
 
+// CreateFrom reads content from rd and stores it as type t. Returned is the
+// storage ID. If the blob is already present, returns ErrAlreadyPresent and
+// the blob's ID.
+func (b *Local) CreateFrom(t Type, rd io.Reader) (ID, error) {
+	// TODO: make sure that tempfile is removed upon error
+
+	// check hash while writing
+	hr := NewHashingReader(rd, newHash())
+
+	// create tempfile in backend
+	file, err := b.tempFile()
+	if err != nil {
+		return nil, err
+	}
+
+	// write data to tempfile
+	_, err = io.Copy(file, hr)
+	if err != nil {
+		return nil, err
+	}
+
+	err = file.Close()
+	if err != nil {
+		return nil, err
+	}
+
+	// get ID
+	id := ID(hr.Sum(nil))
+
+	// check for duplicate ID
+	res, err := b.Test(t, id)
+	if err != nil {
+		return nil, arrar.Annotate(err, "test for presence")
+	}
+
+	if res {
+		return id, ErrAlreadyPresent
+	}
+
+	// rename file
+	err = b.renameFile(file, t, id)
+	if err != nil {
+		return nil, err
+	}
+
+	return id, nil
+}
+
 // Construct path for given Type and ID.
 func (b *Local) filename(t Type, id ID) string {
 	return filepath.Join(b.dirname(t, id), id.String())
@@ -256,6 +305,23 @@ func (b *Local) Get(t Type, id ID) ([]byte, error) {
 	return buf, nil
 }
 
+// GetReader returns a reader that yields the content stored under the given
+// ID. The content is not verified. The reader should be closed after draining
+// it.
+func (b *Local) GetReader(t Type, id ID) (io.ReadCloser, error) {
+	if id == nil {
+		return nil, errors.New("unable to load nil ID")
+	}
+
+	// try to open file
+	file, err := os.Open(b.filename(t, id))
+	if err != nil {
+		return nil, err
+	}
+
+	return file, nil
+}
+
 // Test returns true if a blob of the given type and ID exists in the backend.
 func (b *Local) Test(t Type, id ID) (bool, error) {
 	// try to open file
diff --git a/backend/local_test.go b/backend/local_test.go
index 52e0688b5..2639d8ba3 100644
--- a/backend/local_test.go
+++ b/backend/local_test.go
@@ -1,6 +1,7 @@
 package backend_test
 
 import (
+	"bytes"
 	"flag"
 	"fmt"
 	"io/ioutil"
@@ -80,8 +81,34 @@ func testBackend(b *backend.Local, t *testing.T) {
 
 			// compare content
 			equals(t, test.data, string(buf))
+
+			// compare content again via stream function
+			rd, err := b.GetReader(tpe, id)
+			ok(t, err)
+			buf, err = ioutil.ReadAll(rd)
+			ok(t, err)
+			equals(t, test.data, string(buf))
 		}
 
+		// test stream functions with first file
+		test := TestStrings[0]
+		id, err := backend.ParseID(test.id)
+		ok(t, err)
+
+		// create with reader, should return error
+		_, err = b.CreateFrom(tpe, bytes.NewReader([]byte(test.data)))
+		assert(t, err == backend.ErrAlreadyPresent,
+			"expected ErrAlreadyPresent, got %v", err)
+
+		// remove and recreate
+		err = b.Remove(tpe, id)
+		ok(t, err)
+
+		id2, err := b.CreateFrom(tpe, bytes.NewReader([]byte(test.data)))
+		ok(t, err)
+		assert(t, id.Equal(id2),
+			"expected id %v, got %v", id, id2)
+
 		// list items
 		IDs := backend.IDs{}