feat: Add ReadFrom support

branch.go

@@ -2,8 +2,9 @@ package stream
 
 import (
 	"bytes"
+	"errors"
 	"fmt"
-	"log"
+	"io"
 	"sync"
 
 	"github.com/beyondstorage/go-storage/v4/types"
@@ -50,6 +51,37 @@ func (br *Branch) Write(idx uint64, data []byte) (n int64, err error) {
 	return n, nil
 }
 
+func (br *Branch) ReadFrom(r io.Reader) (n int64, err error) {
+	// Use 4mb as chunk.
+	chunkSize := int64(4 * 1024 * 1024)
+	idx := uint64(0)
+	n = 0
+
+	for {
+		p := formatPath(br.id, idx)
+
+		// Update index.
+		idx++
+
+		size, err := br.s.upper.Write(p, r, chunkSize)
+		// No matter we read success or not, we both need to send data.
+		n += size
+		br.wg.Add(1)
+		br.s.ch <- op{
+			br:   br,
+			size: size,
+		}
+
+		if err != nil && errors.Is(err, io.EOF) {
+			break
+		}
+		if err != nil {
+			return n, err
+		}
+	}
+	return n, nil
+}
+
 func (br *Branch) Complete() (err error) {
 	br.wg.Add(1)
 	br.s.ch <- op{
@@ -89,7 +121,6 @@ func (br *Branch) persist(size int64, done bool) {
 
 	// All data has been persisted, return directly.
 	if br.currentSize.Load()-br.persistedSize == 0 {
-		log.Printf("skip for no data")
 		br.wg.Done()
 		return
 	}

multipart.go

@@ -1,6 +1,7 @@
 package stream
 
 import (
+	"fmt"
 	"github.com/beyondstorage/go-storage/v4/types"
 )
 
@@ -28,20 +29,21 @@ func (br *Branch) persistViaMultipart(start, end uint64, size int64, partNumber
 
 	r, err := br.s.read(br.id, start, end)
 	if err != nil {
-		br.s.errch <- err
+		br.s.errch <- fmt.Errorf("pipe read id %d from %d to %d: %w",
+			br.id, start, end, err)
 		return
 	}
 	defer func() {
 		err = r.Close()
 		if err != nil {
-			br.s.errch <- err
+			br.s.errch <- fmt.Errorf("close pipe reader: %w", err)
 			return
 		}
 	}()
 
 	_, part, err := br.s.underMultipart.WriteMultipart(br.object, r, size, partNumber)
 	if err != nil {
-		br.s.errch <- err
+		br.s.errch <- fmt.Errorf("write multipart: %w", err)
 		return
 	}
 

stream.go

@@ -3,7 +3,6 @@ package stream
 import (
 	"fmt"
 	"io"
-	"log"
 	"sync"
 
 	"github.com/beyondstorage/go-storage/v4/types"
@@ -75,13 +74,13 @@ func (s *Stream) read(id, start, end uint64) (r io.ReadCloser, err error) {
 			p := formatPath(id, i)
 			_, err := s.upper.Read(p, w)
 			if err != nil {
-				s.errch <- err
+				s.errch <- fmt.Errorf("pipe read from %s: %w", p, err)
 				return
 			}
 		}
 		err := w.Close()
 		if err != nil {
-			s.errch <- err
+			s.errch <- fmt.Errorf("close pipe writter: %w", err)
 			return
 		}
 	}()
@@ -94,7 +93,6 @@ func (s *Stream) delete(id, start, end uint64) (err error) {
 		p := formatPath(id, i)
 		err = s.upper.Delete(p)
 		if err != nil {
-			log.Printf("object %s not found", p)
 			return
 		}
 	}

tests/go.mod

@@ -3,10 +3,11 @@ module github.com/beyondstorage/go-stream/tests
 go 1.16
 
 require (
-	github.com/beyondstorage/go-service-memory v0.1.0
+	github.com/beyondstorage/go-service-memory v0.2.0
 	github.com/beyondstorage/go-service-s3/v2 v2.3.0
-	github.com/beyondstorage/go-storage/v4 v4.4.0
+	github.com/beyondstorage/go-storage/v4 v4.5.0
 	github.com/beyondstorage/go-stream v0.1.0
+	github.com/google/uuid v1.3.0
 	github.com/stretchr/testify v1.7.0
 )
 

tests/go.sum

@@ -7,14 +7,16 @@ github.com/aws/aws-sdk-go v1.40.1/go.mod h1:585smgzpB/KqRA+K3y/NL/oYRqQvpNJYvLm+
 github.com/beyondstorage/go-endpoint v1.0.1 h1:F8x2dGLMu9je6g7zPbKoxCXDlug97K26SeCx7KEHgyg=
 github.com/beyondstorage/go-endpoint v1.0.1/go.mod h1:P2hknaGrziOJJKySv/XnAiVw/d3v12/LZu2gSxEx4nM=
 github.com/beyondstorage/go-integration-test/v4 v4.2.0/go.mod h1:jLyYWSGUjQRH7U1HdaLbXE5sxBgqrtK73q+Q7PGIuSs=
-github.com/beyondstorage/go-service-memory v0.1.0 h1:12beKpRiY15wfTRyLsoQWXKELtWri4C2NprUQmX1AcU=
-github.com/beyondstorage/go-service-memory v0.1.0/go.mod h1:GWAXoWnfevG/Weu+UhLzHMJRoZ2ylL3pdYb2kA/w++E=
+github.com/beyondstorage/go-integration-test/v4 v4.3.0/go.mod h1:HKgzemQZpxoHBL49JYEUnLTb5eteUhzcvmmPL7EDT/Y=
+github.com/beyondstorage/go-service-memory v0.2.0 h1:VYIgkKc25XpudmBozu5FJWjuW0+WhBn/uAb+RDhxpKQ=
+github.com/beyondstorage/go-service-memory v0.2.0/go.mod h1:XQwDQ2AwNxt2ImUHgPyPedHcQxUUgxY3EwUEJUvfY1o=
 github.com/beyondstorage/go-service-s3/v2 v2.3.0 h1:mCdCuGmQ/26A1Xhj9tnFcRy3bXf1h5MT/TyFrJW8CAI=
 github.com/beyondstorage/go-service-s3/v2 v2.3.0/go.mod h1:B4gLy8g/uhjkLkVymkRES/tIKH2UhB0zmkzqzGaN5LU=
 github.com/beyondstorage/go-storage/v4 v4.3.0/go.mod h1:0fdcRCzLKMQe7Ve4zPlyTGgoPYwuINiV79Gx9tCt9tQ=
 github.com/beyondstorage/go-storage/v4 v4.3.2/go.mod h1:8FHjTUFuwLl/mmIGpOL9g3RTZPyye7vneFN/JkRj5Tg=
-github.com/beyondstorage/go-storage/v4 v4.4.0 h1:sWURraKFjNR4qpwthr45cAGOIx6EOLrrJcz6su4Je30=
 github.com/beyondstorage/go-storage/v4 v4.4.0/go.mod h1:mc9VzBImjXDg1/1sLfta2MJH79elfM6m47ZZvZ+q/Uw=
+github.com/beyondstorage/go-storage/v4 v4.5.0 h1:cWkX1A+yVLldEpj/Ad9capev9K1fbVIVcRvwqWsMgpw=
+github.com/beyondstorage/go-storage/v4 v4.5.0/go.mod h1:mc9VzBImjXDg1/1sLfta2MJH79elfM6m47ZZvZ+q/Uw=
 github.com/dave/dst v0.26.2 h1:lnxLAKI3tx7MgLNVDirFCsDTlTG9nKTk7GcptKcWSwY=
 github.com/dave/dst v0.26.2/go.mod h1:UMDJuIRPfyUCC78eFuB+SV/WI8oDeyFDvM/JR6NI3IU=
 github.com/dave/gopackages v0.0.0-20170318123100-46e7023ec56e/go.mod h1:i00+b/gKdIDIxuLDFob7ustLAVqhsZRk2qVZrArELGQ=

tests/stream_test.go

@@ -2,41 +2,15 @@ package tests
 
 import (
 	"bytes"
-	"os"
+	"math/rand"
 	"testing"
 
-	_ "github.com/beyondstorage/go-service-memory"
-	_ "github.com/beyondstorage/go-service-s3/v2"
-	"github.com/beyondstorage/go-storage/v4/services"
+	"github.com/google/uuid"
 	"github.com/stretchr/testify/assert"
-
-	"github.com/beyondstorage/go-stream"
 )
 
-func TestStream(t *testing.T) {
-	upperStore, err := services.NewStoragerFromString(os.Getenv("STREAM_UPPER_STORAGE"))
-	if err != nil {
-		t.Fatalf("init upper storage: %v", err)
-	}
-	underStore, err := services.NewStoragerFromString(os.Getenv("STREAM_UNDER_STORAGE"))
-	if err != nil {
-		t.Fatalf("init under storage: %v", err)
-	}
-
-	s, err := stream.NewWithConfig(&stream.Config{
-		Upper:         upperStore,
-		Under:         underStore,
-		PersistMethod: stream.PersistMethodMultipart,
-	})
-	if err != nil {
-		t.Fatalf("init stream: %v", err)
-	}
-	go s.Serve()
-	go func() {
-		for v := range s.Errors() {
-			t.Logf("got error: %v", v)
-		}
-	}()
+func TestStreamWrite(t *testing.T) {
+	s, _, under := setup(t)
 
 	// A bytes with 1k.
 	bs := bytes.Repeat([]byte{'a'}, 1024)
@@ -51,7 +25,9 @@ func TestStream(t *testing.T) {
 
 	for _, v := range cases {
 		t.Run(v.name, func(t *testing.T) {
-			br, err := s.StartBranch(1, "abc")
+			name := uuid.NewString()
+
+			br, err := s.StartBranch(rand.Uint64(), name)
 			if err != nil {
 				t.Fatal(err)
 			}
@@ -61,13 +37,14 @@ func TestStream(t *testing.T) {
 					t.Fatal(err)
 				}
 			}
+
 			err = br.Complete()
 			if err != nil {
 				t.Fatal(err)
 			}
 
 			var actualContent bytes.Buffer
-			_, err = underStore.Read("abc", &actualContent)
+			_, err = under.Read(name, &actualContent)
 			if err != nil {
 				t.Fatal(err)
 			}
@@ -76,3 +53,48 @@ func TestStream(t *testing.T) {
 		})
 	}
 }
+
+func TestStreamReadFrom(t *testing.T) {
+	s, _, under := setup(t)
+
+	cases := []struct {
+		name string
+		size int
+	}{
+		{"1MB", 1024 * 1024},
+		{"16MB", 16 * 1024 * 1024},
+	}
+
+	for _, v := range cases {
+		t.Run(v.name, func(t *testing.T) {
+			name := uuid.NewString()
+
+			// A bytes with size.
+			bs := bytes.Repeat([]byte{'a'}, v.size)
+
+			br, err := s.StartBranch(rand.Uint64(), name)
+			if err != nil {
+				t.Fatal(err)
+			}
+
+			n, err := br.ReadFrom(bytes.NewReader(bs))
+			if err != nil {
+				t.Fatal(err)
+			}
+			assert.Equal(t, n, int64(v.size))
+
+			err = br.Complete()
+			if err != nil {
+				t.Fatal(err)
+			}
+
+			var actualContent bytes.Buffer
+			_, err = under.Read(name, &actualContent)
+			if err != nil {
+				t.Fatal(err)
+			}
+
+			assert.Equal(t, bs, actualContent.Bytes())
+		})
+	}
+}

tests/utils_test.go

@@ -0,0 +1,41 @@
+package tests
+
+import (
+	"os"
+	"testing"
+
+	_ "github.com/beyondstorage/go-service-memory"
+	_ "github.com/beyondstorage/go-service-s3/v2"
+	"github.com/beyondstorage/go-storage/v4/services"
+	"github.com/beyondstorage/go-storage/v4/types"
+
+	"github.com/beyondstorage/go-stream"
+)
+
+func setup(t *testing.T) (s *stream.Stream, upper, under types.Storager) {
+	upperStore, err := services.NewStoragerFromString(os.Getenv("STREAM_UPPER_STORAGE"))
+	if err != nil {
+		t.Fatalf("init upper storage: %v", err)
+	}
+	underStore, err := services.NewStoragerFromString(os.Getenv("STREAM_UNDER_STORAGE"))
+	if err != nil {
+		t.Fatalf("init under storage: %v", err)
+	}
+
+	s, err = stream.NewWithConfig(&stream.Config{
+		Upper:         upperStore,
+		Under:         underStore,
+		PersistMethod: stream.PersistMethodMultipart,
+	})
+	if err != nil {
+		t.Fatalf("init stream: %v", err)
+	}
+	go s.Serve()
+	go func() {
+		for v := range s.Errors() {
+			t.Logf("got error: %v", v)
+		}
+	}()
+
+	return s, upperStore, underStore
+}