Refactor stream helpers

Author: rsora

arduino/utils/stream.go

@@ -0,0 +1,52 @@
+// This file is part of arduino-cli.
+//
+// Copyright 2020 ARDUINO SA (http://www.arduino.cc/)
+//
+// This software is released under the GNU General Public License version 3,
+// which covers the main part of arduino-cli.
+// The terms of this license can be found at:
+// https://www.gnu.org/licenses/gpl-3.0.en.html
+//
+// You can be released from the requirements of the above licenses by purchasing
+// a commercial license. Buying such a license is mandatory if you want to
+// modify or otherwise use the software for commercial activities involving the
+// Arduino software without disclosing the source code of your own applications.
+// To purchase a commercial license, send an email to license@arduino.cc.
+
+package utils
+
+import "io"
+
+// FeedStreamTo creates a pipe to pass data to the writer function.
+// FeedStreamTo returns the io.Writer side of the pipe, on which the user can write data
+func FeedStreamTo(writer func(data []byte)) io.Writer {
+	r, w := io.Pipe()
+	go func() {
+		data := make([]byte, 1024)
+		for {
+			if n, err := r.Read(data); err == nil {
+				writer(data[:n])
+			} else {
+				return
+			}
+		}
+	}()
+	return w
+}
+
+// ConsumeStreamFrom creates a pipe to consume data from the reader function.
+// ConsumeStreamFrom returns the io.Reader side of the pipe, which the user can use to consume the data
+func ConsumeStreamFrom(reader func() ([]byte, error)) io.Reader {
+
+	r, w := io.Pipe()
+	go func() {
+		for {
+			if data, err := reader(); err != nil {
+				return
+			} else if _, err := w.Write(data); err != nil {
+				return
+			}
+		}
+	}()
+	return r
+}

commands/daemon/daemon.go

@@ -19,9 +19,9 @@ package daemon
 
 import (
 	"context"
-	"io"
 	"net/http"
 
+	"github.com/arduino/arduino-cli/arduino/utils"
 	"github.com/arduino/arduino-cli/commands"
 	"github.com/arduino/arduino-cli/commands/board"
 	"github.com/arduino/arduino-cli/commands/compile"
@@ -125,8 +125,8 @@ func (s *ArduinoCoreServerImpl) Version(ctx context.Context, req *rpc.VersionReq
 func (s *ArduinoCoreServerImpl) Compile(req *rpc.CompileReq, stream rpc.ArduinoCore_CompileServer) error {
 	resp, err := compile.Compile(
 		stream.Context(), req,
-		feedStream(func(data []byte) { stream.Send(&rpc.CompileResp{OutStream: data}) }),
-		feedStream(func(data []byte) { stream.Send(&rpc.CompileResp{ErrStream: data}) }),
+		utils.FeedStreamTo(func(data []byte) { stream.Send(&rpc.CompileResp{OutStream: data}) }),
+		utils.FeedStreamTo(func(data []byte) { stream.Send(&rpc.CompileResp{ErrStream: data}) }),
 		false) // set debug to false
 	if err != nil {
 		return err
@@ -222,21 +222,6 @@ func (s *ArduinoCoreServerImpl) Upload(req *rpc.UploadReq, stream rpc.ArduinoCor
 	return stream.Send(resp)
 }
 
-func feedStream(streamer func(data []byte)) io.Writer {
-	r, w := io.Pipe()
-	go func() {
-		data := make([]byte, 1024)
-		for {
-			if n, err := r.Read(data); err == nil {
-				streamer(data[:n])
-			} else {
-				return
-			}
-		}
-	}()
-	return w
-}
-
 // LibraryDownload FIXMEDOC
 func (s *ArduinoCoreServerImpl) LibraryDownload(req *rpc.LibraryDownloadReq, stream rpc.ArduinoCore_LibraryDownloadServer) error {
 	resp, err := lib.LibraryDownload(

commands/daemon/debug.go

@@ -17,9 +17,8 @@ package daemon
 
 import (
 	"fmt"
+	"github.com/arduino/arduino-cli/arduino/utils"
 	cmd "github.com/arduino/arduino-cli/commands/debug"
-	"io"
-
 	dbg "github.com/arduino/arduino-cli/rpc/debug"
 )
 
@@ -45,30 +44,15 @@ func (s *DebugService) Debug(stream dbg.Debug_DebugServer) error {
 
 	// launch debug recipe attaching stdin and out to grpc streaming
 	resp, err := cmd.Debug(stream.Context(), req,
-		copyStream(func() ([]byte, error) {
+		utils.ConsumeStreamFrom(func() ([]byte, error) {
 			command, err := stream.Recv()
 			return command.GetData(), err
 		}),
-		feedStream(func(data []byte) {
+		utils.FeedStreamTo(func(data []byte) {
 			stream.Send(&dbg.DebugResp{Data: data})
 		}))
 	if err != nil {
 		return (err)
 	}
 	return stream.Send(resp)
 }
-
-func copyStream(streamIn func() ([]byte, error)) io.Reader {
-
-	r, w := io.Pipe()
-	go func() {
-		for {
-			if data, err := streamIn(); err != nil {
-				return
-			} else if _, err := w.Write(data); err != nil {
-				return
-			}
-		}
-	}()
-	return r
-}