Skip to content

Commit 49bfbc5

Browse files
cmagliecmaglie
committed
Use callback instead of channels
1 parent 91cd5a6 commit 49bfbc5

File tree

3 files changed

+150
-165
lines changed

3 files changed

+150
-165
lines changed

‎internal/update/apt/service.go‎

Lines changed: 52 additions & 56 deletions
Original file line numberDiff line numberDiff line change
@@ -73,75 +73,71 @@ func (s *Service) ListUpgradablePackages(ctx context.Context, matcher func(updat
7373
// UpgradePackages upgrades the specified packages using the `apt-get upgrade` command.
7474
// It publishes events to subscribers during the upgrade process.
7575
// It returns an error if the upgrade is already in progress or if the upgrade command fails.
76-
func (s *Service) UpgradePackages(ctx context.Context, names []string) (<-chan update.Event, error) {
76+
func (s *Service) UpgradePackages(ctx context.Context, names []string, eventCB func(update.Event)) error {
7777
if !s.lock.TryLock() {
78-
return nil, update.ErrOperationAlreadyInProgress
78+
return update.ErrOperationAlreadyInProgress
7979
}
80-
eventsCh := make(chan update.Event, 100)
8180

82-
go func() {
83-
defer s.lock.Unlock()
84-
defer close(eventsCh)
81+
defer s.lock.Unlock()
8582

86-
// At the end of the upgrade, always try to restart the services (that need it).
87-
// This makes sure key services are restarted even if an error happens in the upgrade steps (for examples container images download).
88-
defer func() {
89-
eventsCh <- update.NewDataEvent(update.RestartEvent, "Upgrade completed. Restarting ...")
83+
// At the end of the upgrade, always try to restart the services (that need it).
84+
// This makes sure key services are restarted even if an error happens in the upgrade steps (for examples container images download).
85+
defer func() {
86+
eventCB(update.NewDataEvent(update.RestartEvent, "Upgrade completed. Restarting ..."))
9087

91-
err := restartServices(ctx)
92-
if err != nil {
93-
eventsCh <- update.NewErrorEvent(fmt.Errorf("error restarting services after upgrade: %w", err))
94-
return
95-
}
96-
}()
97-
98-
eventsCh <- update.NewDataEvent(update.StartEvent, "Upgrade is starting")
99-
stream := runUpgradeCommand(ctx, names)
100-
for line, err := range stream {
101-
if err != nil {
102-
eventsCh <- update.NewErrorEvent(fmt.Errorf("error running upgrade command: %w", err))
103-
return
104-
}
105-
eventsCh <- update.NewDataEvent(update.UpgradeLineEvent, line)
88+
err := restartServices(ctx)
89+
if err != nil {
90+
eventCB(update.NewErrorEvent(fmt.Errorf("error restarting services after upgrade: %w", err)))
91+
return
10692
}
93+
}()
10794

108-
eventsCh <- update.NewDataEvent(update.StartEvent, "apt cleaning cache is starting")
109-
for line, err := range runAptCleanCommand(ctx) {
110-
if err != nil {
111-
eventsCh <- update.NewErrorEvent(fmt.Errorf("error running apt clean command: %w", err))
112-
return
113-
}
114-
eventsCh <- update.NewDataEvent(update.UpgradeLineEvent, line)
95+
eventCB(update.NewDataEvent(update.StartEvent, "Upgrade is starting"))
96+
stream := runUpgradeCommand(ctx, names)
97+
for line, err := range stream {
98+
if err != nil {
99+
eventCB(update.NewErrorEvent(fmt.Errorf("error running upgrade command: %w", err)))
100+
return nil
115101
}
102+
eventCB(update.NewDataEvent(update.UpgradeLineEvent, line))
103+
}
116104

117-
eventsCh <- update.NewDataEvent(update.UpgradeLineEvent, "Stop and destroy docker containers and images ....")
118-
streamCleanup := cleanupDockerContainers(ctx)
119-
for line, err := range streamCleanup {
120-
if err != nil {
121-
// TODO: maybe we should retun an error or a better feedback to the user?
122-
// currently, we just log the error and continue considenring not blocking
123-
slog.Warn("Error stopping and destroying docker containers", "error", err)
124-
} else {
125-
eventsCh <- update.NewDataEvent(update.UpgradeLineEvent, line)
126-
}
105+
eventCB(update.NewDataEvent(update.StartEvent, "apt cleaning cache is starting"))
106+
for line, err := range runAptCleanCommand(ctx) {
107+
if err != nil {
108+
eventCB(update.NewErrorEvent(fmt.Errorf("error running apt clean command: %w", err)))
109+
return nil
127110
}
111+
eventCB(update.NewDataEvent(update.UpgradeLineEvent, line))
112+
}
128113

129-
// TODO: Remove this workaround once docker image versions are no longer hardcoded in arduino-app-cli.
130-
// Tracking issue: https://github.com/arduino/arduino-app-cli/issues/600
131-
// Currently, we need to launch `arduino-app-cli system init` to pull the latest docker images because
132-
// the version of the docker images are hardcoded in the (new downloaded) version of the arduino-app-cli.
133-
eventsCh <- update.NewDataEvent(update.UpgradeLineEvent, "Pulling the latest docker images ...")
134-
streamDocker := pullDockerImages(ctx)
135-
for line, err := range streamDocker {
136-
if err != nil {
137-
eventsCh <- update.NewErrorEvent(fmt.Errorf("error pulling docker images: %w", err))
138-
return
139-
}
140-
eventsCh <- update.NewDataEvent(update.UpgradeLineEvent, line)
114+
eventCB(update.NewDataEvent(update.UpgradeLineEvent, "Stop and destroy docker containers and images ...."))
115+
streamCleanup := cleanupDockerContainers(ctx)
116+
for line, err := range streamCleanup {
117+
if err != nil {
118+
// TODO: maybe we should retun an error or a better feedback to the user?
119+
// currently, we just log the error and continue considenring not blocking
120+
slog.Warn("Error stopping and destroying docker containers", "error", err)
121+
} else {
122+
eventCB(update.NewDataEvent(update.UpgradeLineEvent, line))
141123
}
142-
}()
124+
}
143125

144-
return eventsCh, nil
126+
// TODO: Remove this workaround once docker image versions are no longer hardcoded in arduino-app-cli.
127+
// Tracking issue: https://github.com/arduino/arduino-app-cli/issues/600
128+
// Currently, we need to launch `arduino-app-cli system init` to pull the latest docker images because
129+
// the version of the docker images are hardcoded in the (new downloaded) version of the arduino-app-cli.
130+
eventCB(update.NewDataEvent(update.UpgradeLineEvent, "Pulling the latest docker images ..."))
131+
streamDocker := pullDockerImages(ctx)
132+
for line, err := range streamDocker {
133+
if err != nil {
134+
eventCB(update.NewErrorEvent(fmt.Errorf("error pulling docker images: %w", err)))
135+
return nil
136+
}
137+
eventCB(update.NewDataEvent(update.UpgradeLineEvent, line))
138+
}
139+
140+
return nil
145141
}
146142

147143
// runDpkgConfigureCommand is need in case an upgrade was interrupted in the middle

‎internal/update/arduino/arduino.go‎

Lines changed: 94 additions & 98 deletions
Original file line numberDiff line numberDiff line change
@@ -134,129 +134,125 @@ func (a *ArduinoPlatformUpdater) ListUpgradablePackages(ctx context.Context, _ f
134134
}
135135

136136
// UpgradePackages implements ServiceUpdater.
137-
func (a *ArduinoPlatformUpdater) UpgradePackages(ctx context.Context, names []string) (<-chan update.Event, error) {
137+
func (a *ArduinoPlatformUpdater) UpgradePackages(ctx context.Context, names []string, eventCB func(update.Event)) error {
138138
if !a.lock.TryLock() {
139-
return nil, update.ErrOperationAlreadyInProgress
139+
return update.ErrOperationAlreadyInProgress
140140
}
141-
eventsCh := make(chan update.Event, 100)
142141

143142
downloadProgressCB := func(curr *rpc.DownloadProgress) {
144143
data := helpers.ArduinoCLIDownloadProgressToString(curr)
145144
slog.Debug("Download progress", slog.String("download_progress", data))
146-
eventsCh <- update.NewDataEvent(update.UpgradeLineEvent, data)
145+
eventCB(update.NewDataEvent(update.UpgradeLineEvent, data))
147146
}
148147
taskProgressCB := func(msg *rpc.TaskProgress) {
149148
data := helpers.ArduinoCLITaskProgressToString(msg)
150149
slog.Debug("Task progress", slog.String("task_progress", data))
151-
eventsCh <- update.NewDataEvent(update.UpgradeLineEvent, data)
150+
eventCB(update.NewDataEvent(update.UpgradeLineEvent, data))
152151
}
153152

154-
go func() {
155-
defer a.lock.Unlock()
156-
defer close(eventsCh)
153+
defer a.lock.Unlock()
157154

158-
eventsCh <- update.NewDataEvent(update.StartEvent, "Upgrade is starting")
155+
eventCB(update.NewDataEvent(update.StartEvent, "Upgrade is starting"))
159156

160-
logrus.SetLevel(logrus.ErrorLevel) // Reduce the log level of arduino-cli
161-
srv := commands.NewArduinoCoreServer()
157+
logrus.SetLevel(logrus.ErrorLevel) // Reduce the log level of arduino-cli
158+
srv := commands.NewArduinoCoreServer()
162159

163-
if err := setConfig(ctx, srv); err != nil {
164-
eventsCh <- update.NewErrorEvent(fmt.Errorf("error setting config: %w", err))
165-
return
160+
if err := setConfig(ctx, srv); err != nil {
161+
eventCB(update.NewErrorEvent(fmt.Errorf("error setting config: %w", err)))
162+
return nil
163+
}
164+
165+
var inst *rpc.Instance
166+
if resp, err := srv.Create(ctx, &rpc.CreateRequest{}); err != nil {
167+
eventCB(update.NewErrorEvent(fmt.Errorf("error creating arduino-cli instance: %w", err)))
168+
return nil
169+
} else {
170+
inst = resp.GetInstance()
171+
}
172+
defer func() {
173+
_, err := srv.CleanDownloadCacheDirectory(ctx, &rpc.CleanDownloadCacheDirectoryRequest{})
174+
if err != nil {
175+
slog.Error("Error cleaning cache directory", slog.Any("error", err))
166176
}
177+
_, _ = srv.Destroy(ctx, &rpc.DestroyRequest{Instance: inst})
178+
}()
167179

168-
var inst *rpc.Instance
169-
if resp, err := srv.Create(ctx, &rpc.CreateRequest{}); err != nil {
170-
eventsCh <- update.NewErrorEvent(fmt.Errorf("error creating arduino-cli instance: %w", err))
171-
return
172-
} else {
173-
inst = resp.GetInstance()
180+
{
181+
stream, _ := commands.UpdateIndexStreamResponseToCallbackFunction(ctx, downloadProgressCB)
182+
if err := srv.UpdateIndex(&rpc.UpdateIndexRequest{Instance: inst}, stream); err != nil {
183+
eventCB(update.NewErrorEvent(fmt.Errorf("error updating index: %w", err)))
184+
return nil
174185
}
175-
defer func() {
176-
_, err := srv.CleanDownloadCacheDirectory(ctx, &rpc.CleanDownloadCacheDirectoryRequest{})
177-
if err != nil {
178-
slog.Error("Error cleaning cache directory", slog.Any("error", err))
179-
}
180-
_, _ = srv.Destroy(ctx, &rpc.DestroyRequest{Instance: inst})
181-
}()
182-
183-
{
184-
stream, _ := commands.UpdateIndexStreamResponseToCallbackFunction(ctx, downloadProgressCB)
185-
if err := srv.UpdateIndex(&rpc.UpdateIndexRequest{Instance: inst}, stream); err != nil {
186-
eventsCh <- update.NewErrorEvent(fmt.Errorf("error updating index: %w", err))
187-
return
188-
}
189-
if err := srv.Init(&rpc.InitRequest{Instance: inst}, commands.InitStreamResponseToCallbackFunction(ctx, nil)); err != nil {
190-
eventsCh <- update.NewErrorEvent(fmt.Errorf("error initializing instance: %w", err))
191-
return
192-
}
186+
if err := srv.Init(&rpc.InitRequest{Instance: inst}, commands.InitStreamResponseToCallbackFunction(ctx, nil)); err != nil {
187+
eventCB(update.NewErrorEvent(fmt.Errorf("error initializing instance: %w", err)))
188+
return nil
193189
}
190+
}
194191

195-
stream, respCB := commands.PlatformUpgradeStreamResponseToCallbackFunction(
196-
ctx,
197-
downloadProgressCB,
198-
taskProgressCB,
199-
)
200-
if err := srv.PlatformUpgrade(
201-
&rpc.PlatformUpgradeRequest{
202-
Instance: inst,
203-
PlatformPackage: "arduino",
204-
Architecture: "zephyr",
205-
SkipPostInstall: false,
206-
SkipPreUninstall: false,
207-
},
208-
stream,
209-
); err != nil {
210-
var alreadyPresent *cmderrors.PlatformAlreadyAtTheLatestVersionError
211-
if errors.As(err, &alreadyPresent) {
212-
eventsCh <- update.NewDataEvent(update.UpgradeLineEvent, alreadyPresent.Error())
213-
return
214-
}
215-
216-
var notFound *cmderrors.PlatformNotFoundError
217-
if !errors.As(err, &notFound) {
218-
eventsCh <- update.NewErrorEvent(fmt.Errorf("error upgrading platform: %w", err))
219-
return
220-
}
221-
// If the platform is not found, we will try to install it
222-
err := srv.PlatformInstall(
223-
&rpc.PlatformInstallRequest{
224-
Instance: inst,
225-
PlatformPackage: "arduino",
226-
Architecture: "zephyr",
227-
},
228-
commands.PlatformInstallStreamResponseToCallbackFunction(
229-
ctx,
230-
downloadProgressCB,
231-
taskProgressCB,
232-
),
233-
)
234-
if err != nil {
235-
eventsCh <- update.NewErrorEvent(fmt.Errorf("error installing platform: %w", err))
236-
return
237-
}
238-
} else if respCB().GetPlatform() == nil {
239-
eventsCh <- update.NewErrorEvent(fmt.Errorf("platform upgrade failed"))
240-
return
192+
stream, respCB := commands.PlatformUpgradeStreamResponseToCallbackFunction(
193+
ctx,
194+
downloadProgressCB,
195+
taskProgressCB,
196+
)
197+
if err := srv.PlatformUpgrade(
198+
&rpc.PlatformUpgradeRequest{
199+
Instance: inst,
200+
PlatformPackage: "arduino",
201+
Architecture: "zephyr",
202+
SkipPostInstall: false,
203+
SkipPreUninstall: false,
204+
},
205+
stream,
206+
); err != nil {
207+
var alreadyPresent *cmderrors.PlatformAlreadyAtTheLatestVersionError
208+
if errors.As(err, &alreadyPresent) {
209+
eventCB(update.NewDataEvent(update.UpgradeLineEvent, alreadyPresent.Error()))
210+
return nil
241211
}
242212

243-
cbw := orchestrator.NewCallbackWriter(func(line string) {
244-
eventsCh <- update.NewDataEvent(update.UpgradeLineEvent, line)
245-
})
246-
247-
err := srv.BurnBootloader(
248-
&rpc.BurnBootloaderRequest{
249-
Instance: inst,
250-
Fqbn: "arduino:zephyr:unoq",
251-
Programmer: "jlink",
213+
var notFound *cmderrors.PlatformNotFoundError
214+
if !errors.As(err, &notFound) {
215+
eventCB(update.NewErrorEvent(fmt.Errorf("error upgrading platform: %w", err)))
216+
return nil
217+
}
218+
// If the platform is not found, we will try to install it
219+
err := srv.PlatformInstall(
220+
&rpc.PlatformInstallRequest{
221+
Instance: inst,
222+
PlatformPackage: "arduino",
223+
Architecture: "zephyr",
252224
},
253-
commands.BurnBootloaderToServerStreams(ctx, cbw, cbw),
225+
commands.PlatformInstallStreamResponseToCallbackFunction(
226+
ctx,
227+
downloadProgressCB,
228+
taskProgressCB,
229+
),
254230
)
255231
if err != nil {
256-
eventsCh <- update.NewErrorEvent(fmt.Errorf("error burning bootloader: %w", err))
257-
return
232+
eventCB(update.NewErrorEvent(fmt.Errorf("error installing platform: %w", err)))
233+
return nil
258234
}
259-
}()
235+
} else if respCB().GetPlatform() == nil {
236+
eventCB(update.NewErrorEvent(fmt.Errorf("platform upgrade failed")))
237+
return nil
238+
}
239+
240+
cbw := orchestrator.NewCallbackWriter(func(line string) {
241+
eventCB(update.NewDataEvent(update.UpgradeLineEvent, line))
242+
})
260243

261-
return eventsCh, nil
244+
err := srv.BurnBootloader(
245+
&rpc.BurnBootloaderRequest{
246+
Instance: inst,
247+
Fqbn: "arduino:zephyr:unoq",
248+
Programmer: "jlink",
249+
},
250+
commands.BurnBootloaderToServerStreams(ctx, cbw, cbw),
251+
)
252+
if err != nil {
253+
eventCB(update.NewErrorEvent(fmt.Errorf("error burning bootloader: %w", err)))
254+
return nil
255+
}
256+
257+
return nil
262258
}

‎internal/update/update.go‎

Lines changed: 4 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -46,7 +46,7 @@ type UpgradablePackage struct {
4646

4747
type ServiceUpdater interface {
4848
ListUpgradablePackages(ctx context.Context, matcher func(UpgradablePackage) bool) ([]UpgradablePackage, error)
49-
UpgradePackages(ctx context.Context, names []string) (<-chan Event, error)
49+
UpgradePackages(ctx context.Context, names []string, eventCB func(Event)) error
5050
}
5151

5252
type Manager struct {
@@ -131,28 +131,21 @@ func (m *Manager) UpgradePackages(ctx context.Context, pkgs []UpgradablePackage)
131131

132132
go func() {
133133
defer m.lock.Unlock()
134+
134135
// We are launching on purpose the update sequentially. The reason is that
135136
// the deb pkgs restart the orchestrator, and if we run in parallel the
136137
// update of the cores we will end up with inconsistent state, or
137138
// we need to re run the upgrade because the orchestrator interrupted
138139
// in the middle the upgrade of the cores.
139-
arduinoEvents, err := m.arduinoPlatformUpdateService.UpgradePackages(ctx, arduinoPlatform)
140-
if err != nil {
140+
if err := m.arduinoPlatformUpdateService.UpgradePackages(ctx, arduinoPlatform, m.broadcast); err != nil {
141141
m.broadcast(NewErrorEvent(fmt.Errorf("failed to upgrade Arduino packages: %w", err)))
142142
return
143143
}
144-
for e := range arduinoEvents {
145-
m.broadcast(e)
146-
}
147144

148-
aptEvents, err := m.debUpdateService.UpgradePackages(ctx, debPkgs)
149-
if err != nil {
145+
if err := m.debUpdateService.UpgradePackages(ctx, debPkgs, m.broadcast); err != nil {
150146
m.broadcast(NewErrorEvent(fmt.Errorf("failed to upgrade APT packages: %w", err)))
151147
return
152148
}
153-
for e := range aptEvents {
154-
m.broadcast(e)
155-
}
156149

157150
m.broadcast(NewDataEvent(DoneEvent, "Update completed"))
158151
}()

0 commit comments

Comments
 (0)