Apply stream-send synchronization in Compile grpc call

commands/daemon/daemon.go

@@ -202,16 +202,17 @@ func (s *ArduinoCoreServerImpl) LoadSketch(ctx context.Context, req *rpc.LoadSke
 
 // Compile FIXMEDOC
 func (s *ArduinoCoreServerImpl) Compile(req *rpc.CompileRequest, stream rpc.ArduinoCoreService_CompileServer) error {
-	outStream := feedStreamTo(func(data []byte) { stream.Send(&rpc.CompileResponse{OutStream: data}) })
-	errStream := feedStreamTo(func(data []byte) { stream.Send(&rpc.CompileResponse{ErrStream: data}) })
+	syncSend := NewSynchronizedSend(stream.Send)
+	outStream := feedStreamTo(func(data []byte) { syncSend.Send(&rpc.CompileResponse{OutStream: data}) })
+	errStream := feedStreamTo(func(data []byte) { syncSend.Send(&rpc.CompileResponse{ErrStream: data}) })
 	compileResp, compileErr := compile.Compile(
 		stream.Context(), req, outStream, errStream,
-		func(p *rpc.TaskProgress) { stream.Send(&rpc.CompileResponse{Progress: p}) })
+		func(p *rpc.TaskProgress) { syncSend.Send(&rpc.CompileResponse{Progress: p}) })
 	outStream.Close()
 	errStream.Close()
 	var compileRespSendErr error
 	if compileResp != nil {
-		compileRespSendErr = stream.Send(compileResp)
+		compileRespSendErr = syncSend.Send(compileResp)
 	}
 	if compileErr != nil {
 		return convertErrorToRPCStatus(compileErr)

commands/daemon/stream.go

@@ -98,3 +98,26 @@ func consumeStreamFrom(reader func() ([]byte, error)) io.Reader {
 	}()
 	return r
 }
+
+// SynchronizedSender is a sender function with an extra protection for
+// concurrent writes, if multiple threads call the Send method they will
+// be blocked and serialized.
+type SynchronizedSender[T any] struct {
+	lock          sync.Mutex
+	protectedSend func(T) error
+}
+
+// Send the message using the underlyng stream.
+func (s *SynchronizedSender[T]) Send(value T) error {
+	s.lock.Lock()
+	err := s.protectedSend(value)
+	s.lock.Unlock()
+	return err
+}
+
+// NewSynchronizedSend takes a Send function and wraps it in a SynchronizedSender
+func NewSynchronizedSend[T any](send func(T) error) *SynchronizedSender[T] {
+	return &SynchronizedSender[T]{
+		protectedSend: send,
+	}
+}