Inlining methods in ArduinoCoreServiceImpl (part 4)

This commit feature a refactoring of the BoardListWatch command, now the
gRPC stream is converted into a channel by leveraging the new stub
boardListWatchServer that may be generic-ized to be reused for other
streaming calls.

Author: cmaglie

Diff for: commands/service.go

@@ -18,7 +18,6 @@ package commands
 import (
 	"context"
 	"errors"
-	"fmt"
 	"io"
 	"sync/atomic"
 
@@ -42,37 +41,6 @@ type arduinoCoreServerImpl struct {
 	versionString string
 }
 
-// BoardSearch exposes to the gRPC interface the board search command
-func (s *arduinoCoreServerImpl) BoardSearch(ctx context.Context, req *rpc.BoardSearchRequest) (*rpc.BoardSearchResponse, error) {
-	return BoardSearch(ctx, req)
-}
-
-// BoardListWatch FIXMEDOC
-func (s *arduinoCoreServerImpl) BoardListWatch(req *rpc.BoardListWatchRequest, stream rpc.ArduinoCoreService_BoardListWatchServer) error {
-	syncSend := NewSynchronizedSend(stream.Send)
-	if req.GetInstance() == nil {
-		err := fmt.Errorf(tr("no instance specified"))
-		syncSend.Send(&rpc.BoardListWatchResponse{
-			EventType: "error",
-			Error:     err.Error(),
-		})
-		return err
-	}
-
-	eventsChan, err := BoardListWatch(stream.Context(), req)
-	if err != nil {
-		return err
-	}
-
-	for event := range eventsChan {
-		if err := syncSend.Send(event); err != nil {
-			logrus.Infof("sending board watch message: %v", err)
-		}
-	}
-
-	return nil
-}
-
 // Destroy FIXMEDOC
 func (s *arduinoCoreServerImpl) Destroy(ctx context.Context, req *rpc.DestroyRequest) (*rpc.DestroyResponse, error) {
 	return Destroy(ctx, req)

Diff for: commands/service_board_list.go

@@ -25,6 +25,7 @@ import (
 	"regexp"
 	"sort"
 	"strings"
+	"sync"
 	"time"
 
 	"github.com/arduino/arduino-cli/commands/cmderrors"
@@ -38,6 +39,7 @@ import (
 	"github.com/arduino/go-properties-orderedmap"
 	discovery "github.com/arduino/pluggable-discovery-protocol-handler/v2"
 	"github.com/sirupsen/logrus"
+	"google.golang.org/grpc/metadata"
 )
 
 var (
@@ -262,29 +264,72 @@ func hasMatchingBoard(b *rpc.DetectedPort, fqbnFilter *cores.FQBN) bool {
 	return false
 }
 
-// BoardListWatch returns a channel that receives boards connection and disconnection events.
-func BoardListWatch(ctx context.Context, req *rpc.BoardListWatchRequest) (<-chan *rpc.BoardListWatchResponse, error) {
+type boardListWatchServer struct {
+	ctx      context.Context
+	respChan chan<- *rpc.BoardListWatchResponse
+	respLock sync.Mutex
+}
+
+func NewBoardListWatchServerToChan(ctx context.Context, target chan<- *rpc.BoardListWatchResponse) rpc.ArduinoCoreService_BoardListWatchServer {
+	w := &boardListWatchServer{ctx: ctx, respChan: target}
+	go func() {
+		<-ctx.Done()
+
+		w.respLock.Lock()
+		close(w.respChan)
+		w.respChan = nil
+		w.respLock.Unlock()
+	}()
+	return w
+}
+
+func (w *boardListWatchServer) Send(resp *rpc.BoardListWatchResponse) error {
+	w.respLock.Lock()
+	if w.respChan != nil {
+		w.respChan <- resp
+	}
+	w.respLock.Unlock()
+	return nil
+}
+
+func (w *boardListWatchServer) Context() context.Context     { return w.ctx }
+func (w *boardListWatchServer) RecvMsg(m any) error          { return errors.New("not implemented") }
+func (w *boardListWatchServer) SendHeader(metadata.MD) error { return errors.New("not implemented") }
+func (w *boardListWatchServer) SendMsg(m any) error          { return errors.New("not implemented") }
+func (w *boardListWatchServer) SetHeader(metadata.MD) error  { return errors.New("not implemented") }
+func (w *boardListWatchServer) SetTrailer(metadata.MD)       {}
+
+// BoardListWatch FIXMEDOC
+func (s *arduinoCoreServerImpl) BoardListWatch(req *rpc.BoardListWatchRequest, stream rpc.ArduinoCoreService_BoardListWatchServer) error {
+	syncSend := NewSynchronizedSend(stream.Send)
+	if req.GetInstance() == nil {
+		err := fmt.Errorf(tr("no instance specified"))
+		syncSend.Send(&rpc.BoardListWatchResponse{
+			EventType: "error",
+			Error:     err.Error(),
+		})
+		return err
+	}
+
 	pme, release, err := instances.GetPackageManagerExplorer(req.GetInstance())
 	if err != nil {
-		return nil, err
+		return err
 	}
 	defer release()
 	dm := pme.DiscoveryManager()
 
 	watcher, err := dm.Watch()
 	if err != nil {
-		return nil, err
+		return err
 	}
 
 	go func() {
-		<-ctx.Done()
+		<-stream.Context().Done()
 		logrus.Trace("closed watch")
 		watcher.Close()
 	}()
 
-	outChan := make(chan *rpc.BoardListWatchResponse)
 	go func() {
-		defer close(outChan)
 		for event := range watcher.Feed() {
 			port := &rpc.DetectedPort{
 				Port: rpc.DiscoveryPortToRPC(event.Port),
@@ -298,13 +343,13 @@ func BoardListWatch(ctx context.Context, req *rpc.BoardListWatchRequest) (<-chan
 				}
 				port.MatchingBoards = boards
 			}
-			outChan <- &rpc.BoardListWatchResponse{
+			stream.Send(&rpc.BoardListWatchResponse{
 				EventType: event.Type,
 				Port:      port,
 				Error:     boardsError,
-			}
+			})
 		}
 	}()
 
-	return outChan, nil
+	return nil
 }

Diff for: commands/service_board_search.go

@@ -29,7 +29,7 @@ import (
 // Boards are searched in all platforms, including those in the index that are not yet
 // installed. Note that platforms that are not installed don't include boards' FQBNs.
 // If no search argument is used all boards are returned.
-func BoardSearch(ctx context.Context, req *rpc.BoardSearchRequest) (*rpc.BoardSearchResponse, error) {
+func (s *arduinoCoreServerImpl) BoardSearch(ctx context.Context, req *rpc.BoardSearchRequest) (*rpc.BoardSearchResponse, error) {
 	pme, release, err := instances.GetPackageManagerExplorer(req.GetInstance())
 	if err != nil {
 		return nil, err

Diff for: internal/cli/arguments/fqbn.go

@@ -85,7 +85,7 @@ func CalculateFQBNAndPort(portArgs *Port, fqbnArg *Fqbn, instance *rpc.Instance,
 		return fqbn, port
 	}
 
-	port, err := portArgs.GetPort(instance, defaultAddress, defaultProtocol)
+	port, err := portArgs.GetPort(instance, srv, defaultAddress, defaultProtocol)
 	if err != nil {
 		feedback.Fatal(tr("Error getting port metadata: %v", err), feedback.ErrGeneric)
 	}

Diff for: internal/cli/arguments/port.go

@@ -56,12 +56,12 @@ func (p *Port) AddToCommand(cmd *cobra.Command, srv rpc.ArduinoCoreServiceServer
 // This method allows will bypass the discoveries if:
 // - a nil instance is passed: in this case the plain port and protocol arguments are returned (even if empty)
 // - a protocol is specified: in this case the discoveries are not needed to autodetect the protocol.
-func (p *Port) GetPortAddressAndProtocol(instance *rpc.Instance, defaultAddress, defaultProtocol string) (string, string, error) {
+func (p *Port) GetPortAddressAndProtocol(instance *rpc.Instance, srv rpc.ArduinoCoreServiceServer, defaultAddress, defaultProtocol string) (string, string, error) {
 	if p.protocol != "" || instance == nil {
 		return p.address, p.protocol, nil
 	}
 
-	port, err := p.GetPort(instance, defaultAddress, defaultProtocol)
+	port, err := p.GetPort(instance, srv, defaultAddress, defaultProtocol)
 	if err != nil {
 		return "", "", err
 	}
@@ -70,8 +70,7 @@ func (p *Port) GetPortAddressAndProtocol(instance *rpc.Instance, defaultAddress,
 
 // GetPort returns the Port obtained by parsing command line arguments.
 // The extra metadata for the ports is obtained using the pluggable discoveries.
-func (p *Port) GetPort(instance *rpc.Instance, defaultAddress, defaultProtocol string) (*rpc.Port, error) {
-
+func (p *Port) GetPort(instance *rpc.Instance, srv rpc.ArduinoCoreServiceServer, defaultAddress, defaultProtocol string) (*rpc.Port, error) {
 	address := p.address
 	protocol := p.protocol
 	if address == "" && (defaultAddress != "" || defaultProtocol != "") {
@@ -91,7 +90,12 @@ func (p *Port) GetPort(instance *rpc.Instance, defaultAddress, defaultProtocol s
 
 	ctx, cancel := context.WithCancel(context.Background())
 	defer cancel()
-	watcher, err := commands.BoardListWatch(ctx, &rpc.BoardListWatchRequest{Instance: instance})
+
+	watcher := make(chan *rpc.BoardListWatchResponse, 1)
+	err := srv.BoardListWatch(
+		&rpc.BoardListWatchRequest{Instance: instance},
+		commands.NewBoardListWatchServerToChan(ctx, watcher))
+
 	if err != nil {
 		return nil, err
 	}

Diff for: internal/cli/board/attach.go

@@ -45,7 +45,7 @@ func initAttachCommand(srv rpc.ArduinoCoreServiceServer) *cobra.Command {
 			if len(args) > 0 {
 				sketchPath = args[0]
 			}
-			runAttachCommand(sketchPath, &port, fqbn.String(), &programmer)
+			runAttachCommand(srv, sketchPath, &port, fqbn.String(), &programmer)
 		},
 	}
 	fqbn.AddToCommand(attachCommand, srv)
@@ -55,10 +55,10 @@ func initAttachCommand(srv rpc.ArduinoCoreServiceServer) *cobra.Command {
 	return attachCommand
 }
 
-func runAttachCommand(path string, port *arguments.Port, fqbn string, programmer *arguments.Programmer) {
+func runAttachCommand(srv rpc.ArduinoCoreServiceServer, path string, port *arguments.Port, fqbn string, programmer *arguments.Programmer) {
 	sketchPath := arguments.InitSketchPath(path)
 
-	portAddress, portProtocol, _ := port.GetPortAddressAndProtocol(nil, "", "")
+	portAddress, portProtocol, _ := port.GetPortAddressAndProtocol(nil, srv, "", "")
 	newDefaults, err := commands.SetSketchDefaults(context.Background(), &rpc.SetSketchDefaultsRequest{
 		SketchPath:          sketchPath.String(),
 		DefaultFqbn:         fqbn,

Diff for: internal/cli/board/board.go

@@ -39,7 +39,7 @@ func NewCommand(srv rpc.ArduinoCoreServiceServer) *cobra.Command {
 	boardCommand.AddCommand(initDetailsCommand(srv))
 	boardCommand.AddCommand(initListCommand(srv))
 	boardCommand.AddCommand(initListAllCommand(srv))
-	boardCommand.AddCommand(initSearchCommand())
+	boardCommand.AddCommand(initSearchCommand(srv))
 
 	return boardCommand
 }

Diff for: internal/cli/board/list.go

@@ -63,7 +63,7 @@ func runListCommand(srv rpc.ArduinoCoreServiceServer, watch bool, timeout int64,
 	logrus.Info("Executing `arduino-cli board list`")
 
 	if watch {
-		watchList(inst)
+		watchList(inst, srv)
 		return
 	}
 
@@ -88,8 +88,11 @@ func runListCommand(srv rpc.ArduinoCoreServiceServer, watch bool, timeout int64,
 	feedback.PrintResult(listResult{result.NewDetectedPorts(ports)})
 }
 
-func watchList(inst *rpc.Instance) {
-	eventsChan, err := commands.BoardListWatch(context.Background(), &rpc.BoardListWatchRequest{Instance: inst})
+func watchList(inst *rpc.Instance, srv rpc.ArduinoCoreServiceServer) {
+	eventsChan := make(chan *rpc.BoardListWatchResponse, 1)
+	err := srv.BoardListWatch(
+		&rpc.BoardListWatchRequest{Instance: inst},
+		commands.NewBoardListWatchServerToChan(context.Background(), eventsChan))
 	if err != nil {
 		feedback.Fatal(tr("Error detecting boards: %v", err), feedback.ErrNetwork)
 	}

Diff for: internal/cli/board/search.go

@@ -22,7 +22,6 @@ import (
 	"sort"
 	"strings"
 
-	"github.com/arduino/arduino-cli/commands"
 	"github.com/arduino/arduino-cli/internal/cli/feedback"
 	"github.com/arduino/arduino-cli/internal/cli/feedback/result"
 	"github.com/arduino/arduino-cli/internal/cli/feedback/table"
@@ -32,7 +31,7 @@ import (
 	"github.com/spf13/cobra"
 )
 
-func initSearchCommand() *cobra.Command {
+func initSearchCommand(srv rpc.ArduinoCoreServiceServer) *cobra.Command {
 	var searchCommand = &cobra.Command{
 		Use:   fmt.Sprintf("search [%s]", tr("boardname")),
 		Short: tr("Search for a board in the Boards Manager."),
@@ -41,18 +40,20 @@ func initSearchCommand() *cobra.Command {
 			"  " + os.Args[0] + " board search\n" +
 			"  " + os.Args[0] + " board search zero",
 		Args: cobra.ArbitraryArgs,
-		Run:  runSearchCommand,
+		Run: func(cmd *cobra.Command, args []string) {
+			runSearchCommand(srv, args)
+		},
 	}
 	searchCommand.Flags().BoolVarP(&showHiddenBoard, "show-hidden", "a", false, tr("Show also boards marked as 'hidden' in the platform"))
 	return searchCommand
 }
 
-func runSearchCommand(cmd *cobra.Command, args []string) {
+func runSearchCommand(srv rpc.ArduinoCoreServiceServer, args []string) {
 	inst := instance.CreateAndInit()
 
 	logrus.Info("Executing `arduino-cli board search`")
 
-	res, err := commands.BoardSearch(context.Background(), &rpc.BoardSearchRequest{
+	res, err := srv.BoardSearch(context.Background(), &rpc.BoardSearchRequest{
 		Instance:            inst,
 		SearchArgs:          strings.Join(args, " "),
 		IncludeHiddenBoards: showHiddenBoard,

Diff for: internal/cli/burnbootloader/burnbootloader.go

@@ -71,7 +71,7 @@ func runBootloaderCommand(srv rpc.ArduinoCoreServiceServer) {
 	logrus.Info("Executing `arduino-cli burn-bootloader`")
 
 	// We don't need a Sketch to upload a board's bootloader
-	discoveryPort, err := port.GetPort(instance, "", "")
+	discoveryPort, err := port.GetPort(instance, srv, "", "")
 	if err != nil {
 		feedback.Fatal(tr("Error during Upload: %v", err), feedback.ErrGeneric)
 	}

Diff for: internal/cli/debug/debug_check.go

@@ -55,7 +55,7 @@ func runDebugCheckCommand(srv rpc.ArduinoCoreServiceServer, portArgs *arguments.
 	instance := instance.CreateAndInit()
 	logrus.Info("Executing `arduino-cli debug`")
 
-	port, err := portArgs.GetPort(instance, "", "")
+	port, err := portArgs.GetPort(instance, srv, "", "")
 	if err != nil {
 		feedback.FatalError(err, feedback.ErrBadArgument)
 	}

Diff for: internal/cli/monitor/monitor.go

@@ -140,7 +140,7 @@ func runMonitorCmd(
 		fqbn, _ = portArgs.DetectFQBN(inst, srv)
 	}
 
-	portAddress, portProtocol, err := portArgs.GetPortAddressAndProtocol(inst, defaultPort, defaultProtocol)
+	portAddress, portProtocol, err := portArgs.GetPortAddressAndProtocol(inst, srv, defaultPort, defaultProtocol)
 	if err != nil {
 		feedback.FatalError(err, feedback.ErrGeneric)
 	}