Use callback instead of channels

cmaglie · cmaglie · commit 07fbbfbdf105 · 2025-12-23T14:31:42.000+01:00
diff --git a/internal/update/apt/service.go b/internal/update/apt/service.go
@@ -73,75 +73,73 @@ func (s *Service) ListUpgradablePackages(ctx context.Context, matcher func(updat
 // UpgradePackages upgrades the specified packages using the `apt-get upgrade` command.
 // It publishes events to subscribers during the upgrade process.
 // It returns an error if the upgrade is already in progress or if the upgrade command fails.
-func (s *Service) UpgradePackages(ctx context.Context, names []string) (<-chan update.Event, error) {
+func (s *Service) UpgradePackages(ctx context.Context, names []string, eventCB func(update.Event)) error {
 	if !s.lock.TryLock() {
-		return nil, update.ErrOperationAlreadyInProgress
+		return update.ErrOperationAlreadyInProgress
 	}
-	eventsCh := make(chan update.Event, 100)
 
 	go func() {
 		defer s.lock.Unlock()
-		defer close(eventsCh)
 
 		// At the end of the upgrade, always try to restart the services (that need it).
 		// This makes sure key services are restarted even if an error happens in the upgrade steps (for examples container images download).
 		defer func() {
-			eventsCh <- update.NewDataEvent(update.RestartEvent, "Upgrade completed. Restarting ...")
+			eventCB(update.NewDataEvent(update.RestartEvent, "Upgrade completed. Restarting ..."))
 
 			err := restartServices(ctx)
 			if err != nil {
-				eventsCh <- update.NewErrorEvent(fmt.Errorf("error restarting services after upgrade: %w", err))
+				eventCB(update.NewErrorEvent(fmt.Errorf("error restarting services after upgrade: %w", err)))
 				return
 			}
 		}()
 
-		eventsCh <- update.NewDataEvent(update.StartEvent, "Upgrade is starting")
+		eventCB(update.NewDataEvent(update.StartEvent, "Upgrade is starting"))
 		stream := runUpgradeCommand(ctx, names)
 		for line, err := range stream {
 			if err != nil {
-				eventsCh <- update.NewErrorEvent(fmt.Errorf("error running upgrade command: %w", err))
+				eventCB(update.NewErrorEvent(fmt.Errorf("error running upgrade command: %w", err)))
 				return
 			}
-			eventsCh <- update.NewDataEvent(update.UpgradeLineEvent, line)
+			eventCB(update.NewDataEvent(update.UpgradeLineEvent, line))
 		}
 
-		eventsCh <- update.NewDataEvent(update.StartEvent, "apt cleaning cache is starting")
+		eventCB(update.NewDataEvent(update.StartEvent, "apt cleaning cache is starting"))
 		for line, err := range runAptCleanCommand(ctx) {
 			if err != nil {
-				eventsCh <- update.NewErrorEvent(fmt.Errorf("error running apt clean command: %w", err))
+				eventCB(update.NewErrorEvent(fmt.Errorf("error running apt clean command: %w", err)))
 				return
 			}
-			eventsCh <- update.NewDataEvent(update.UpgradeLineEvent, line)
+			eventCB(update.NewDataEvent(update.UpgradeLineEvent, line))
 		}
 
-		eventsCh <- update.NewDataEvent(update.UpgradeLineEvent, "Stop and destroy docker containers and images ....")
+		eventCB(update.NewDataEvent(update.UpgradeLineEvent, "Stop and destroy docker containers and images ...."))
 		streamCleanup := cleanupDockerContainers(ctx)
 		for line, err := range streamCleanup {
 			if err != nil {
 				// TODO: maybe we should retun an error or a better feedback to the user?
 				// currently, we just log the error and continue considenring not blocking
 				slog.Warn("Error stopping and destroying docker containers", "error", err)
 			} else {
-				eventsCh <- update.NewDataEvent(update.UpgradeLineEvent, line)
+				eventCB(update.NewDataEvent(update.UpgradeLineEvent, line))
 			}
 		}
 
 		// TODO: Remove this workaround once docker image versions are no longer hardcoded in arduino-app-cli.
 		// Tracking issue: https://github.com/arduino/arduino-app-cli/issues/600
 		// Currently, we need to launch `arduino-app-cli system init` to pull the latest docker images because
 		// the version of the docker images are hardcoded in the (new downloaded) version of the arduino-app-cli.
-		eventsCh <- update.NewDataEvent(update.UpgradeLineEvent, "Pulling the latest docker images ...")
+		eventCB(update.NewDataEvent(update.UpgradeLineEvent, "Pulling the latest docker images ..."))
 		streamDocker := pullDockerImages(ctx)
 		for line, err := range streamDocker {
 			if err != nil {
-				eventsCh <- update.NewErrorEvent(fmt.Errorf("error pulling docker images: %w", err))
+				eventCB(update.NewErrorEvent(fmt.Errorf("error pulling docker images: %w", err)))
 				return
 			}
-			eventsCh <- update.NewDataEvent(update.UpgradeLineEvent, line)
+			eventCB(update.NewDataEvent(update.UpgradeLineEvent, line))
 		}
 	}()
 
-	return eventsCh, nil
+	return nil
 }
 
 // runDpkgConfigureCommand is need in case an upgrade was interrupted in the middle
diff --git a/internal/update/arduino/arduino.go b/internal/update/arduino/arduino.go
@@ -134,40 +134,38 @@ func (a *ArduinoPlatformUpdater) ListUpgradablePackages(ctx context.Context, _ f
 }
 
 // UpgradePackages implements ServiceUpdater.
-func (a *ArduinoPlatformUpdater) UpgradePackages(ctx context.Context, names []string) (<-chan update.Event, error) {
+func (a *ArduinoPlatformUpdater) UpgradePackages(ctx context.Context, names []string, eventCB func(update.Event)) error {
 	if !a.lock.TryLock() {
-		return nil, update.ErrOperationAlreadyInProgress
+		return update.ErrOperationAlreadyInProgress
 	}
-	eventsCh := make(chan update.Event, 100)
 
 	downloadProgressCB := func(curr *rpc.DownloadProgress) {
 		data := helpers.ArduinoCLIDownloadProgressToString(curr)
 		slog.Debug("Download progress", slog.String("download_progress", data))
-		eventsCh <- update.NewDataEvent(update.UpgradeLineEvent, data)
+		eventCB(update.NewDataEvent(update.UpgradeLineEvent, data))
 	}
 	taskProgressCB := func(msg *rpc.TaskProgress) {
 		data := helpers.ArduinoCLITaskProgressToString(msg)
 		slog.Debug("Task progress", slog.String("task_progress", data))
-		eventsCh <- update.NewDataEvent(update.UpgradeLineEvent, data)
+		eventCB(update.NewDataEvent(update.UpgradeLineEvent, data))
 	}
 
 	go func() {
 		defer a.lock.Unlock()
-		defer close(eventsCh)
 
-		eventsCh <- update.NewDataEvent(update.StartEvent, "Upgrade is starting")
+		eventCB(update.NewDataEvent(update.StartEvent, "Upgrade is starting"))
 
 		logrus.SetLevel(logrus.ErrorLevel) // Reduce the log level of arduino-cli
 		srv := commands.NewArduinoCoreServer()
 
 		if err := setConfig(ctx, srv); err != nil {
-			eventsCh <- update.NewErrorEvent(fmt.Errorf("error setting config: %w", err))
+			eventCB(update.NewErrorEvent(fmt.Errorf("error setting config: %w", err)))
 			return
 		}
 
 		var inst *rpc.Instance
 		if resp, err := srv.Create(ctx, &rpc.CreateRequest{}); err != nil {
-			eventsCh <- update.NewErrorEvent(fmt.Errorf("error creating arduino-cli instance: %w", err))
+			eventCB(update.NewErrorEvent(fmt.Errorf("error creating arduino-cli instance: %w", err)))
 			return
 		} else {
 			inst = resp.GetInstance()
@@ -183,11 +181,11 @@ func (a *ArduinoPlatformUpdater) UpgradePackages(ctx context.Context, names []st
 		{
 			stream, _ := commands.UpdateIndexStreamResponseToCallbackFunction(ctx, downloadProgressCB)
 			if err := srv.UpdateIndex(&rpc.UpdateIndexRequest{Instance: inst}, stream); err != nil {
-				eventsCh <- update.NewErrorEvent(fmt.Errorf("error updating index: %w", err))
+				eventCB(update.NewErrorEvent(fmt.Errorf("error updating index: %w", err)))
 				return
 			}
 			if err := srv.Init(&rpc.InitRequest{Instance: inst}, commands.InitStreamResponseToCallbackFunction(ctx, nil)); err != nil {
-				eventsCh <- update.NewErrorEvent(fmt.Errorf("error initializing instance: %w", err))
+				eventCB(update.NewErrorEvent(fmt.Errorf("error initializing instance: %w", err)))
 				return
 			}
 		}
@@ -209,13 +207,13 @@ func (a *ArduinoPlatformUpdater) UpgradePackages(ctx context.Context, names []st
 		); err != nil {
 			var alreadyPresent *cmderrors.PlatformAlreadyAtTheLatestVersionError
 			if errors.As(err, &alreadyPresent) {
-				eventsCh <- update.NewDataEvent(update.UpgradeLineEvent, alreadyPresent.Error())
+				eventCB(update.NewDataEvent(update.UpgradeLineEvent, alreadyPresent.Error()))
 				return
 			}
 
 			var notFound *cmderrors.PlatformNotFoundError
 			if !errors.As(err, &notFound) {
-				eventsCh <- update.NewErrorEvent(fmt.Errorf("error upgrading platform: %w", err))
+				eventCB(update.NewErrorEvent(fmt.Errorf("error upgrading platform: %w", err)))
 				return
 			}
 			// If the platform is not found, we will try to install it
@@ -232,16 +230,16 @@ func (a *ArduinoPlatformUpdater) UpgradePackages(ctx context.Context, names []st
 				),
 			)
 			if err != nil {
-				eventsCh <- update.NewErrorEvent(fmt.Errorf("error installing platform: %w", err))
+				eventCB(update.NewErrorEvent(fmt.Errorf("error installing platform: %w", err)))
 				return
 			}
 		} else if respCB().GetPlatform() == nil {
-			eventsCh <- update.NewErrorEvent(fmt.Errorf("platform upgrade failed"))
+			eventCB(update.NewErrorEvent(fmt.Errorf("platform upgrade failed")))
 			return
 		}
 
 		cbw := orchestrator.NewCallbackWriter(func(line string) {
-			eventsCh <- update.NewDataEvent(update.UpgradeLineEvent, line)
+			eventCB(update.NewDataEvent(update.UpgradeLineEvent, line))
 		})
 
 		err := srv.BurnBootloader(
@@ -253,10 +251,10 @@ func (a *ArduinoPlatformUpdater) UpgradePackages(ctx context.Context, names []st
 			commands.BurnBootloaderToServerStreams(ctx, cbw, cbw),
 		)
 		if err != nil {
-			eventsCh <- update.NewErrorEvent(fmt.Errorf("error burning bootloader: %w", err))
+			eventCB(update.NewErrorEvent(fmt.Errorf("error burning bootloader: %w", err)))
 			return
 		}
 	}()
 
-	return eventsCh, nil
+	return nil
 }
diff --git a/internal/update/update.go b/internal/update/update.go
@@ -46,7 +46,7 @@ type UpgradablePackage struct {
 
 type ServiceUpdater interface {
 	ListUpgradablePackages(ctx context.Context, matcher func(UpgradablePackage) bool) ([]UpgradablePackage, error)
-	UpgradePackages(ctx context.Context, names []string) (<-chan Event, error)
+	UpgradePackages(ctx context.Context, names []string, eventCB func(Event)) error
 }
 
 type Manager struct {
@@ -136,23 +136,15 @@ func (m *Manager) UpgradePackages(ctx context.Context, pkgs []UpgradablePackage)
 		// update of the cores we will end up with inconsistent state, or
 		// we need to re run the upgrade because the orchestrator interrupted
 		// in the middle the upgrade of the cores.
-		arduinoEvents, err := m.arduinoPlatformUpdateService.UpgradePackages(ctx, arduinoPlatform)
-		if err != nil {
+		if err := m.arduinoPlatformUpdateService.UpgradePackages(ctx, arduinoPlatform, m.broadcast); err != nil {
 			m.broadcast(NewErrorEvent(fmt.Errorf("failed to upgrade Arduino packages: %w", err)))
 			return
 		}
-		for e := range arduinoEvents {
-			m.broadcast(e)
-		}
 
-		aptEvents, err := m.debUpdateService.UpgradePackages(ctx, debPkgs)
-		if err != nil {
+		if err := m.debUpdateService.UpgradePackages(ctx, debPkgs, m.broadcast); err != nil {
 			m.broadcast(NewErrorEvent(fmt.Errorf("failed to upgrade APT packages: %w", err)))
 			return
 		}
-		for e := range aptEvents {
-			m.broadcast(e)
-		}
 
 		m.broadcast(NewDataEvent(DoneEvent, "Update completed"))
 	}()
