Stream optional interfaces

backends/fs/fs.go

@@ -3,6 +3,7 @@ package fs
 import (
 	"context"
 	"fmt"
+	"io"
 	"os"
 	"path/filepath"
 	"sort"
@@ -89,6 +90,25 @@ func (b *Backend) Delete(ctx context.Context, name string) error {
 	return err
 }
 
+// NewReader provides an optimized way to read from named file.
+func (b *Backend) NewReader(ctx context.Context, name string) (io.ReadCloser, error) {
+	if !allowedName(name) {
+		return nil, os.ErrPermission
+	}
+	fullPath := filepath.Join(b.rootPath, name)
+	return os.Open(fullPath)
+}
+
+
+// NewWriter provides an optimized way to write to a file.
+func (b *Backend) NewWriter(ctx context.Context, name string) (io.WriteCloser, error) {
+	if !allowedName(name) {
+		return nil, os.ErrPermission
+	}
+	fullPath := filepath.Join(b.rootPath, name)
+	return os.OpenFile(fullPath, os.O_CREATE|os.O_WRONLY|os.O_TRUNC, 0644)
+}
+
 func allowedName(name string) bool {
 	// TODO: Make shared and test for rejection
 	if strings.Contains(name, "/") {

backends/s3/s3.go

@@ -346,6 +346,27 @@ func New(ctx context.Context, opt Options) (*Backend, error) {
 	return b, nil
 }
 
+// NewReader satisfies ReaderStorage and provides an optimized, streamed way
+// to fetch a blob from S3 server.
+func (b *Backend) NewReader(ctx context.Context, name string) (io.ReadCloser, error) {
+	obj, err := b.client.GetObject(ctx, b.opt.Bucket, name, minio.GetObjectOptions{})
+	if err = convertMinioError(err); err != nil {
+		return nil, err
+	}
+	if obj == nil {
+		return nil, os.ErrNotExist
+	}
+
+	info, err := obj.Stat()
+	if err = convertMinioError(err); err != nil {
+		return nil, err
+	}
+	if info.Key != name {
+		return nil, os.ErrNotExist
+	}
+	return obj, nil
+}
+
 // convertMinioError takes an error, possibly a minio.ErrorResponse
 // and turns it into a well known error when possible.
 // If error is not well known, it is returned as is.

stream.go

@@ -0,0 +1,99 @@
+package simpleblob
+
+import (
+	"bytes"
+	"context"
+	"errors"
+	"io"
+	"sync"
+)
+
+// A ReaderStorage is an Interface providing an optimized way to create an io.ReadCloser.
+type ReaderStorage interface {
+	Interface
+	// NewReader returns an io.ReadCloser, allowing stream reading
+	// of named value from the underlying backend.
+	NewReader(context.Context, string) (io.ReadCloser, error)
+}
+
+// A WriterStorage is an Interface providing an optimized way to create an io.WriteCloser.
+type WriterStorage interface {
+	Interface
+	// NewWriter returns an io.WriteCloser, allowing stream writing
+	// to named key in the underlying backend.
+	NewWriter(context.Context, string) (io.WriteCloser, error)
+}
+
+// A reader wraps a backend to satisfy io.ReadCloser.
+type reader struct {
+	st Interface
+	*bytes.Reader
+}
+
+// Close signifies operations on reader are over.
+func (*reader) Close() error {
+	return nil
+}
+
+// NewReader returns an optimized io.ReadCloser for backend if available,
+// else a basic buffered implementation.
+func NewReader(ctx context.Context, st Interface, name string) (io.ReadCloser, error) {
+	if sst, ok := st.(ReaderStorage); ok {
+		return sst.NewReader(ctx, name)
+	}
+	b, err := st.Load(ctx, name)
+	if err != nil {
+		return nil, err
+	}
+	return &reader{st, bytes.NewReader(b)}, nil
+}
+
+// A writer wraps a backend to satisfy io.WriteCloser.
+// The bytes written to it are buffered, then sent to backend when closed.
+type writer struct {
+	st     Interface
+	ctx    context.Context
+	name   string
+	closed bool
+	buf    bytes.Buffer
+	mu     sync.Mutex
+}
+
+var errWClosed = errors.New("WriterStorage closed")
+
+// Write appends p to the data ready to be stored.
+//
+// Content will only be sent to backend when w.Close is called.
+func (w *writer) Write(p []byte) (int, error) {
+	w.mu.Lock()
+	defer w.mu.Unlock()
+	if w.closed {
+		return 0, errWClosed
+	}
+	return w.buf.Write(p)
+}
+
+// Close signifies operations on writer are over.
+// The file is sent to backend when called.
+func (w *writer) Close() error {
+	w.mu.Lock()
+	defer w.mu.Unlock()
+	if w.closed {
+		return errWClosed
+	}
+	w.closed = true
+	return w.st.Store(w.ctx, w.name, w.buf.Bytes())
+}
+
+// NewWriter returns an optimized io.WriteCloser for backend if available,
+// else a basic buffered implementation.
+func NewWriter(ctx context.Context, st Interface, name string) (io.WriteCloser, error) {
+	if sst, ok := st.(WriterStorage); ok {
+		return sst.NewWriter(ctx, name)
+	}
+	return &writer{
+		st:   st,
+		ctx:  ctx,
+		name: name,
+	}, nil
+}

tester/tester.go

@@ -2,6 +2,7 @@ package tester
 
 import (
 	"context"
+	"io"
 	"os"
 	"testing"
 
@@ -51,6 +52,14 @@ func DoBackendTests(t *testing.T, b simpleblob.Interface) {
 	assert.NoError(t, err)
 	assert.Equal(t, data, []byte("foo"))
 
+	// Reader should get the same data as Load
+	r, err := simpleblob.NewReader(ctx, b, "foo-1")
+	assert.NoError(t, err)
+	p, err := io.ReadAll(r)
+	assert.NoError(t, err)
+	assert.Equal(t, data, p)
+	assert.NoError(t, r.Close())
+
 	// Check overwritten data
 	data, err = b.Load(ctx, "bar-1")
 	assert.NoError(t, err)
@@ -68,9 +77,26 @@ func DoBackendTests(t *testing.T, b simpleblob.Interface) {
 	assert.NoError(t, err)
 	assert.Equal(t, data, []byte("foo"))
 
+	// Writer stores data
+	w, err := simpleblob.NewWriter(ctx, b, "fizz")
+	assert.NoError(t, err)
+	assert.NotNil(t, w)
+	buzz := []byte("buzz")
+	n, err := w.Write(buzz)
+	assert.NoError(t, err)
+	assert.NoError(t, w.Close())
+	assert.EqualValues(t, len(buzz), n)
+	data, err = b.Load(ctx, "fizz")
+	assert.NoError(t, err)
+	assert.Equal(t, buzz, data)
+
 	// Load non-existing
 	_, err = b.Load(ctx, "does-not-exist")
 	assert.ErrorIs(t, err, os.ErrNotExist)
+	// With Reader
+	r, err = simpleblob.NewReader(ctx, b, "does-not-exist")
+	assert.ErrorIs(t, err, os.ErrNotExist)
+	assert.Nil(t, r)
 
 	// Delete existing
 	err = b.Delete(ctx, "foo-1")