Move dapr/utils/streams to kit (dapr#68)

* Move dapr/utils/streams to kit

No code changes

Signed-off-by: ItalyPaleAle <[email protected]>

* 💄

Signed-off-by: ItalyPaleAle <[email protected]>

* Lint

Signed-off-by: ItalyPaleAle <[email protected]>

---------

Signed-off-by: ItalyPaleAle <[email protected]>

Author: ItalyPaleAle

go.mod

@@ -10,6 +10,7 @@ require (
 	github.com/mitchellh/mapstructure v1.5.1-0.20220423185008-bf980b35cac4
 	github.com/sirupsen/logrus v1.9.3
 	github.com/stretchr/testify v1.8.4
+	github.com/tidwall/transform v0.0.0-20201103190739-32f242e2dbde
 	golang.org/x/crypto v0.14.0
 	golang.org/x/exp v0.0.0-20231006140011-7918f672742d
 	google.golang.org/genproto/googleapis/rpc v0.0.0-20230822172742-b8732ec3820d

go.sum

@@ -45,6 +45,8 @@ github.com/stretchr/testify v1.7.1/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/
 github.com/stretchr/testify v1.8.0/go.mod h1:yNjHg4UonilssWZ8iaSj1OCr/vHnekPRkoO+kdMU+MU=
 github.com/stretchr/testify v1.8.4 h1:CcVxjf3Q8PM0mHUKJCdn+eZZtm5yQwehR5yeSVQQcUk=
 github.com/stretchr/testify v1.8.4/go.mod h1:sz/lmYIOXD/1dqDmKjjqLyZ2RngseejIcXlSw2iwfAo=
+github.com/tidwall/transform v0.0.0-20201103190739-32f242e2dbde h1:AMNpJRc7P+GTwVbl8DkK2I9I8BBUzNiHuH/tlxrpan0=
+github.com/tidwall/transform v0.0.0-20201103190739-32f242e2dbde/go.mod h1:MvrEmduDUz4ST5pGZ7CABCnOU5f3ZiOAZzT6b1A6nX8=
 github.com/yuin/goldmark v1.4.13/go.mod h1:6yULJ656Px+3vBD8DxQVa3kxgyrAnzto9xy5taEt/CY=
 golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w=
 golang.org/x/crypto v0.0.0-20210921155107-089bfa567519/go.mod h1:GvvjBRRGRdwPK5ydBHafDWAxML/pGHZbMvKqRZ5+Abc=

signals/signals_posix_test.go

@@ -24,7 +24,7 @@ import (
 	"testing"
 	"time"
 
-	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
 )
 
 func TestContext(t *testing.T) {
@@ -35,7 +35,7 @@ func TestContext(t *testing.T) {
 		onlyOneSignalHandler = make(chan struct{})
 
 		ctx := Context()
-		assert.NoError(t, syscall.Kill(syscall.Getpid(), syscall.SIGINT))
+		require.NoError(t, syscall.Kill(syscall.Getpid(), syscall.SIGINT))
 		select {
 		case <-ctx.Done():
 		case <-time.After(1 * time.Second):

streams/limitreadcloser.go

@@ -0,0 +1,81 @@
+/*
+Copyright 2023 The Dapr Authors
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+    http://www.apache.org/licenses/LICENSE-2.0
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package streams
+
+import (
+	"errors"
+	"io"
+)
+
+/*!
+Adapted from the Go 1.18.3 source code
+Copyright 2009 The Go Authors. All rights reserved.
+License: BSD (https://github.com/golang/go/blob/go1.18.3/LICENSE)
+*/
+
+// ErrStreamTooLarge is returned by LimitReadCloser when the stream is too large.
+var ErrStreamTooLarge = errors.New("stream too large")
+
+// LimitReadCloser returns a ReadCloser that reads from r but stops with ErrStreamTooLarge after n bytes.
+func LimitReadCloser(r io.ReadCloser, n int64) io.ReadCloser {
+	return &limitReadCloser{
+		R: r,
+		N: n,
+	}
+}
+
+type limitReadCloser struct {
+	R      io.ReadCloser
+	N      int64
+	closed bool
+}
+
+func (l *limitReadCloser) Read(p []byte) (n int, err error) {
+	if l.N < 0 || l.R == nil {
+		return 0, ErrStreamTooLarge
+	}
+	if len(p) == 0 {
+		return 0, nil
+	}
+	if l.closed {
+		return 0, io.EOF
+	}
+	if int64(len(p)) > (l.N + 1) {
+		p = p[0:(l.N + 1)]
+	}
+	n, err = l.R.Read(p)
+	l.N -= int64(n)
+	if l.N < 0 {
+		// Special case if we just read the "l.N+1" byte
+		if l.N == -1 {
+			n--
+		}
+		if err == nil {
+			err = ErrStreamTooLarge
+		}
+		if !l.closed {
+			l.closed = true
+			l.R.Close()
+		}
+	}
+	return
+}
+
+func (l *limitReadCloser) Close() error {
+	if l.closed {
+		return nil
+	}
+	l.closed = true
+	return l.R.Close()
+}

streams/limitreadcloser_test.go

@@ -0,0 +1,124 @@
+/*
+Copyright 2023 The Dapr Authors
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+    http://www.apache.org/licenses/LICENSE-2.0
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package streams
+
+import (
+	"io"
+	"strings"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+)
+
+func TestLimitReadCloser(t *testing.T) {
+	t.Run("stream shorter than limit", func(t *testing.T) {
+		s := LimitReadCloser(io.NopCloser(strings.NewReader("e ho guardato dentro un'emozione")), 1000)
+		read, err := io.ReadAll(s)
+		require.NoError(t, err)
+		require.Equal(t, "e ho guardato dentro un'emozione", string(read))
+
+		// Reading again should return io.EOF
+		n, err := s.Read(read)
+		require.ErrorIs(t, err, io.EOF)
+		require.Equal(t, 0, n)
+	})
+
+	t.Run("stream has same length as limit", func(t *testing.T) {
+		s := LimitReadCloser(io.NopCloser(strings.NewReader("e ci ho visto dentro tanto amore")), 32)
+		read, err := io.ReadAll(s)
+		require.NoError(t, err)
+		require.Equal(t, "e ci ho visto dentro tanto amore", string(read))
+
+		// Reading again should return io.EOF
+		n, err := s.Read(read)
+		require.ErrorIs(t, err, io.EOF)
+		require.Equal(t, 0, n)
+	})
+
+	t.Run("stream longer than limit", func(t *testing.T) {
+		s := LimitReadCloser(io.NopCloser(strings.NewReader("che ho capito perche' non si comanda al cuore")), 21)
+		read, err := io.ReadAll(s)
+		require.Error(t, err)
+		require.ErrorIs(t, err, ErrStreamTooLarge)
+		require.Equal(t, "che ho capito perche'", string(read))
+
+		// Reading again should return ErrStreamTooLarge again
+		n, err := s.Read(read)
+		require.ErrorIs(t, err, ErrStreamTooLarge)
+		require.Equal(t, 0, n)
+	})
+
+	t.Run("stream longer than limit, read with byte slice", func(t *testing.T) {
+		s := LimitReadCloser(io.NopCloser(strings.NewReader("e va bene cosi'")), 4)
+
+		read := make([]byte, 100)
+		n, err := s.Read(read)
+		require.Error(t, err)
+		require.ErrorIs(t, err, ErrStreamTooLarge)
+		require.Equal(t, "e va", string(read[0:n]))
+
+		// Reading again should return ErrStreamTooLarge again
+		n, err = s.Read(read)
+		require.ErrorIs(t, err, ErrStreamTooLarge)
+		require.Equal(t, 0, n)
+	})
+
+	t.Run("read in two segments", func(t *testing.T) {
+		s := LimitReadCloser(io.NopCloser(strings.NewReader("senza parole")), 9)
+
+		read := make([]byte, 5)
+
+		n, err := s.Read(read)
+		require.NoError(t, err)
+		require.Equal(t, "senza", string(read[0:n]))
+
+		n, err = s.Read(read)
+		require.Error(t, err)
+		require.ErrorIs(t, err, ErrStreamTooLarge)
+		require.Equal(t, " par", string(read[0:n]))
+
+		// Reading again should return ErrStreamTooLarge again
+		n, err = s.Read(read)
+		require.ErrorIs(t, err, ErrStreamTooLarge)
+		require.Equal(t, 0, n)
+	})
+
+	t.Run("close early", func(t *testing.T) {
+		s := LimitReadCloser(io.NopCloser(strings.NewReader("senza parole")), 10)
+
+		// Read 5 bytes then close
+		read := make([]byte, 5)
+		n, err := s.Read(read)
+		require.NoError(t, err)
+		require.Equal(t, "senza", string(read[0:n]))
+
+		// Reading should now return io.EOF
+		err = s.Close()
+		require.NoError(t, err)
+
+		n, err = s.Read(read)
+		require.Error(t, err)
+		require.ErrorIs(t, err, io.EOF)
+		require.Equal(t, 0, n)
+	})
+
+	t.Run("stream is nil", func(t *testing.T) {
+		s := LimitReadCloser(nil, 10)
+
+		// Reading should return ErrStreamTooLarge again
+		n, err := s.Read(make([]byte, 10))
+		require.ErrorIs(t, err, ErrStreamTooLarge)
+		require.Equal(t, 0, n)
+	})
+}

streams/multireadercloser.go

@@ -0,0 +1,99 @@
+/*
+Copyright 2023 The Dapr Authors
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+    http://www.apache.org/licenses/LICENSE-2.0
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package streams
+
+import (
+	"errors"
+	"io"
+	"net/http"
+)
+
+// NewMultiReaderCloser returns a stream that is like io.MultiReader but that can be closed.
+// When the returned stream is closed, it closes the readable streams too, if they implement io.Closer.
+func NewMultiReaderCloser(readers ...io.Reader) *MultiReaderCloser {
+	r := make([]io.Reader, len(readers))
+	copy(r, readers)
+	return &MultiReaderCloser{
+		readers: r,
+	}
+}
+
+/*!
+Adapted from the Go 1.19.3 source code
+Copyright 2009 The Go Authors. All rights reserved.
+License: BSD (https://github.com/golang/go/blob/go1.19.3/LICENSE)
+*/
+
+// MultiReaderCloser is an io.MultiReader that also implements the io.Closer interface to close the readable streams.
+// Readable streams are also closed when we're done reading from them.
+type MultiReaderCloser struct {
+	readers []io.Reader
+}
+
+func (mr *MultiReaderCloser) Read(p []byte) (n int, err error) {
+	for len(mr.readers) > 0 {
+		r := mr.readers[0]
+		n, err = r.Read(p)
+
+		// When reading from a http.Response Body, we may get ErrBodyReadAfterClose if we already read it all
+		// We consider that the same as io.EOF
+		if errors.Is(err, http.ErrBodyReadAfterClose) {
+			err = io.EOF
+			mr.readers = mr.readers[1:]
+		} else if err == io.EOF {
+			if rc, ok := r.(io.Closer); ok {
+				_ = rc.Close()
+			}
+			mr.readers = mr.readers[1:]
+		}
+		if n > 0 || err != io.EOF {
+			if err == io.EOF && len(mr.readers) > 0 {
+				// Don't return EOF yet. More readers remain.
+				err = nil
+			}
+			return
+		}
+	}
+	return 0, io.EOF
+}
+
+func (mr *MultiReaderCloser) WriteTo(w io.Writer) (sum int64, err error) {
+	return mr.writeToWithBuffer(w, make([]byte, 1024*32))
+}
+
+func (mr *MultiReaderCloser) writeToWithBuffer(w io.Writer, buf []byte) (sum int64, err error) {
+	var n int64
+	for i, r := range mr.readers {
+		n, err = io.CopyBuffer(w, r, buf)
+		sum += n
+		if err != nil {
+			mr.readers = mr.readers[i:] // permit resume / retry after error
+			return sum, err
+		}
+		mr.readers[i] = nil // permit early GC
+	}
+	mr.readers = nil
+	return sum, nil
+}
+
+// Close implements io.Closer.
+func (mr *MultiReaderCloser) Close() error {
+	for _, r := range mr.readers {
+		if rc, ok := r.(io.Closer); ok {
+			_ = rc.Close()
+		}
+	}
+	mr.readers = mr.readers[:0]
+	return nil
+}