propertly implement the internal function

Author: lucarin91

cmd/arduino-app-cli/monitor/monitor.go

@@ -21,6 +21,7 @@ import (
 
 	"github.com/spf13/cobra"
 
+	"github.com/arduino/arduino-app-cli/cmd/feedback"
 	"github.com/arduino/arduino-app-cli/internal/monitor"
 )
 
@@ -29,7 +30,11 @@ func NewMonitorCmd() *cobra.Command {
 		Use:   "monitor",
 		Short: "Attach to the microcontroller serial monitor",
 		RunE: func(cmd *cobra.Command, args []string) error {
-			start, err := monitor.NewMonitorHandler(&stdInOutProxy{stdin: os.Stdin, stdout: os.Stdout}) // nolint:forbidigo
+			stdout, _, err := feedback.DirectStreams()
+			if err != nil {
+				return err
+			}
+			start, err := monitor.NewMonitorHandler(&combinedReadWrite{r: os.Stdin, w: stdout}) // nolint:forbidigo
 			if err != nil {
 				return err
 			}
@@ -40,25 +45,19 @@ func NewMonitorCmd() *cobra.Command {
 	}
 }
 
-type stdInOutProxy struct {
-	stdin  io.Reader
-	stdout io.Writer
+type combinedReadWrite struct {
+	r io.Reader
+	w io.Writer
 }
 
-func (s stdInOutProxy) ReadMessage() (int, []byte, error) {
-	var p [1024]byte
-	n, err := s.stdin.Read(p[:])
-	if err != nil {
-		return 0, nil, err
-	}
-	return 1, p[:n], nil
+func (crw *combinedReadWrite) Read(p []byte) (n int, err error) {
+	return crw.r.Read(p)
 }
 
-func (s stdInOutProxy) WriteMessage(messageType int, data []byte) error {
-	_, err := s.stdout.Write(data)
-	return err
+func (crw *combinedReadWrite) Write(p []byte) (n int, err error) {
+	return crw.w.Write(p)
 }
 
-func (s stdInOutProxy) Close() error {
+func (crw *combinedReadWrite) Close() error {
 	return nil
 }

internal/api/handlers/monitor.go

@@ -21,7 +21,6 @@ import (
 	"net"
 	"net/http"
 	"strings"
-	"time"
 
 	"github.com/gorilla/websocket"
 
@@ -40,27 +39,18 @@ func HandleMonitorWS(allowedOrigins []string) http.HandlerFunc {
 	}
 
 	return func(w http.ResponseWriter, r *http.Request) {
-		// Connect to monitor
-		mon, err := net.DialTimeout("tcp", "127.0.0.1:7500", time.Second)
-		if err != nil {
-			slog.Error("Unable to connect to monitor", slog.String("error", err.Error()))
-			render.EncodeResponse(w, http.StatusServiceUnavailable, models.ErrorResponse{Details: "Unable to connect to monitor: " + err.Error()})
-			return
-		}
-
 		// Upgrade the connection to websocket
 		conn, err := upgrader.Upgrade(w, r, nil)
 		if err != nil {
 			// Remember to close monitor connection if websocket upgrade fails.
-			mon.Close()
 
 			slog.Error("Failed to upgrade connection", slog.String("error", err.Error()))
 			render.EncodeResponse(w, http.StatusInternalServerError, map[string]string{"error": "Failed to upgrade connection: " + err.Error()})
 			return
 		}
 
 		// Now the connection is managed by the websocket library, let's move the handlers in the goroutine
-		start, err := monitor.NewMonitorHandler(conn)
+		start, err := monitor.NewMonitorHandler(&wsReadWriteCloser{conn: conn})
 		if err != nil {
 			slog.Error("Unable to start monitor handler", slog.String("error", err.Error()))
 			render.EncodeResponse(w, http.StatusInternalServerError, models.ErrorResponse{Details: "Unable to start monitor handler: " + err.Error()})
@@ -114,3 +104,50 @@ func checkOrigin(origin string, allowedOrigins []string) bool {
 	slog.Error("WebSocket origin check failed", slog.String("origin", origin))
 	return false
 }
+
+type wsReadWriteCloser struct {
+	conn *websocket.Conn
+
+	buff []byte
+}
+
+func (w *wsReadWriteCloser) Read(p []byte) (n int, err error) {
+	if len(w.buff) > 0 {
+		n = copy(p, w.buff)
+		w.buff = w.buff[n:]
+		return n, nil
+	}
+
+	ty, message, err := w.conn.ReadMessage()
+	if err != nil {
+		return 0, mapWebSocketErrors(err)
+	}
+	if ty != websocket.BinaryMessage {
+		return 0, fmt.Errorf("unexpected websocket message type: %d", ty)
+	}
+	w.buff = message
+
+	n = copy(p, w.buff)
+	w.buff = w.buff[n:]
+	return n, nil
+}
+
+func (w *wsReadWriteCloser) Write(p []byte) (n int, err error) {
+	err = w.conn.WriteMessage(websocket.BinaryMessage, p)
+	if err != nil {
+		return 0, mapWebSocketErrors(err)
+	}
+	return len(p), nil
+}
+
+func (w *wsReadWriteCloser) Close() error {
+	w.buff = nil
+	return w.conn.Close()
+}
+
+func mapWebSocketErrors(err error) error {
+	if websocket.IsUnexpectedCloseError(err, websocket.CloseNormalClosure, websocket.CloseGoingAway, websocket.CloseNoStatusReceived, websocket.CloseAbnormalClosure) {
+		return net.ErrClosed
+	}
+	return err
+}

internal/monitor/monitor.go

@@ -17,44 +17,27 @@ package monitor
 
 import (
 	"errors"
-	"fmt"
 	"io"
 	"log/slog"
 	"net"
 	"time"
-
-	"github.com/gorilla/websocket"
 )
 
-type MessageReaderWriter interface {
-	ReadMessage() (messageType int, p []byte, err error)
-	WriteMessage(messageType int, data []byte) error
-	Close() error
-}
+const monitorAddr = "127.0.0.1:7500"
 
-func NewMonitorHandler(rw MessageReaderWriter) (func(), error) {
+func NewMonitorHandler(rw io.ReadWriteCloser) (func(), error) {
 	// Connect to monitor
-	mon, err := net.DialTimeout("tcp", "127.0.0.1:7500", time.Second)
+	monitor, err := net.DialTimeout("tcp", monitorAddr, time.Second)
 	if err != nil {
 		return nil, err
 	}
 
 	return func() {
-		monitorStream(mon, rw)
+		monitorStream(monitor, rw)
 	}, nil
 }
 
-func monitorStream(mon net.Conn, rw MessageReaderWriter) {
-	logWebsocketError := func(msg string, err error) {
-		// Do not log simple close or interruption errors
-		if websocket.IsUnexpectedCloseError(err, websocket.CloseNormalClosure, websocket.CloseGoingAway, websocket.CloseNoStatusReceived, websocket.CloseAbnormalClosure) {
-			if e, ok := err.(*websocket.CloseError); ok {
-				slog.Error(msg, slog.String("closecause", fmt.Sprintf("%d: %s", e.Code, err)))
-			} else {
-				slog.Error(msg, slog.String("error", err.Error()))
-			}
-		}
-	}
+func monitorStream(mon net.Conn, rw io.ReadWriteCloser) {
 	logSocketError := func(msg string, err error) {
 		if !errors.Is(err, net.ErrClosed) && !errors.Is(err, io.EOF) {
 			slog.Error(msg, slog.String("error", err.Error()))
@@ -63,14 +46,15 @@ func monitorStream(mon net.Conn, rw MessageReaderWriter) {
 	go func() {
 		defer mon.Close()
 		defer rw.Close()
+		buff := [1024]byte{}
 		for {
 			// Read from websocket and write to monitor
-			_, msg, err := rw.ReadMessage()
+			n, err := rw.Read(buff[:])
 			if err != nil {
-				logWebsocketError("Error reading from websocket", err)
+				logSocketError("Error reading from websocket", err)
 				return
 			}
-			if _, err := mon.Write(msg); err != nil {
+			if _, err := mon.Write(buff[:n]); err != nil {
 				logSocketError("Error writing to monitor", err)
 				return
 			}
@@ -88,8 +72,8 @@ func monitorStream(mon net.Conn, rw MessageReaderWriter) {
 				return
 			}
 
-			if err := rw.WriteMessage(websocket.BinaryMessage, buff[:n]); err != nil {
-				logWebsocketError("Error writing to websocket", err)
+			if _, err := rw.Write(buff[:n]); err != nil {
+				logSocketError("Error writing to websocket", err)
 				return
 			}
 		}