Skip to content

Commit d6c0a27

Browse files
committed
Simplify command state update methods
Signed-off-by: Didier Wenzek <didier.wenzek@free.fr>
1 parent b48c19c commit d6c0a27

File tree

4 files changed

+40
-24
lines changed

4 files changed

+40
-24
lines changed

crates/core/tedge_agent/src/operation_workflows/actor.rs

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -159,7 +159,7 @@ impl WorkflowActor {
159159
Ok(Some(new_state)) => {
160160
self.persist_command_board().await?;
161161
if new_state.is_init() {
162-
self.process_command_update(new_state.set_log_path(&log_file.path))
162+
self.process_command_update(new_state.with_log_path(&log_file.path))
163163
.await?;
164164
}
165165
}

crates/core/tedge_agent/src/operation_workflows/persist.rs

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -359,7 +359,7 @@ impl WorkflowRepository {
359359
if let Some(current_version) = self.workflows.use_current_version(&operation) {
360360
self.persist_workflow_definition(&operation, &current_version)
361361
.await;
362-
*command = command.clone().set_workflow_version(&current_version);
362+
command.set_workflow_version(&current_version);
363363
}
364364
}
365365
}

crates/core/tedge_api/src/workflow/state.rs

Lines changed: 26 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -186,8 +186,18 @@ impl GenericCommandState {
186186
}
187187
}
188188

189-
pub fn update_with_key_value(self, key: &str, val: &str) -> Self {
190-
self.update_with_json(json!({ key: val }))
189+
pub fn with_key_value(mut self, key: &str, val: &str) -> Self {
190+
self.set_key_value(key, val);
191+
self
192+
}
193+
194+
fn set_key_value(&mut self, key: &str, val: &str) {
195+
if let Some(o) = self.payload.as_object_mut() {
196+
o.insert(key.to_string(), val.into());
197+
}
198+
if key == STATUS {
199+
self.status = val.to_string();
200+
}
191201
}
192202

193203
pub fn get_log_path(&self) -> Option<Utf8PathBuf> {
@@ -197,8 +207,13 @@ impl GenericCommandState {
197207
.map(Utf8PathBuf::from)
198208
}
199209

200-
pub fn set_log_path<P: AsRef<Utf8Path>>(self, path: P) -> Self {
201-
self.update_with_key_value(OP_LOG_PATH_KEY, path.as_ref().as_str())
210+
pub fn with_log_path<P: AsRef<Utf8Path>>(mut self, path: P) -> Self {
211+
self.set_log_path(path);
212+
self
213+
}
214+
215+
pub fn set_log_path<P: AsRef<Utf8Path>>(&mut self, path: P) {
216+
self.set_key_value(OP_LOG_PATH_KEY, path.as_ref().as_str())
202217
}
203218

204219
pub fn workflow_version(&self) -> Option<String> {
@@ -208,8 +223,13 @@ impl GenericCommandState {
208223
.map(|str| str.to_string())
209224
}
210225

211-
pub fn set_workflow_version(self, version: &str) -> Self {
212-
self.update_with_key_value(OP_WORKFLOW_VERSION_KEY, version)
226+
pub fn with_workflow_version(mut self, version: &str) -> Self {
227+
self.set_workflow_version(version);
228+
self
229+
}
230+
231+
pub fn set_workflow_version(&mut self, version: &str) {
232+
self.set_key_value(OP_WORKFLOW_VERSION_KEY, version)
213233
}
214234

215235
/// Update the command state with the outcome of a script

crates/core/tedge_api/src/workflow/supervisor.rs

Lines changed: 12 additions & 16 deletions
Original file line numberDiff line numberDiff line change
@@ -80,7 +80,7 @@ impl WorkflowSupervisor {
8080
self.commands = commands;
8181
self.commands
8282
.iter()
83-
.filter_map(|(t, s)| self.resume_command(t, s))
83+
.filter_map(|(t, s)| self.resume_command(t, s.clone()))
8484
.collect()
8585
}
8686

@@ -136,10 +136,10 @@ impl WorkflowSupervisor {
136136
///
137137
/// Return the current version if any.
138138
pub fn use_current_version(&mut self, operation: &OperationName) -> Option<WorkflowVersion> {
139-
match self.workflows.get_mut(&operation.as_str().into()) {
140-
Some(versions) => versions.use_current_version().cloned(),
141-
None => None,
142-
}
139+
self.workflows
140+
.get_mut(&operation.as_str().into())
141+
.map(WorkflowVersions::use_current_version)?
142+
.cloned()
143143
}
144144

145145
/// Update the state of the command board on reception of a message sent by a peer over MQTT
@@ -162,9 +162,9 @@ impl WorkflowSupervisor {
162162
} else if command_state.is_init() {
163163
// This is a new command request
164164
if let Some(current_version) = workflow_versions.use_current_version() {
165-
let command_state = command_state.set_workflow_version(current_version);
166-
self.commands.insert(command_state.clone())?;
167-
Ok(Some(command_state))
165+
let updated_state = command_state.with_workflow_version(current_version);
166+
self.commands.insert(updated_state.clone())?;
167+
Ok(Some(updated_state))
168168
} else {
169169
return Err(WorkflowExecutionError::DeprecatedOperation {
170170
operation: operation.to_string(),
@@ -287,21 +287,17 @@ impl WorkflowSupervisor {
287287
fn resume_command(
288288
&self,
289289
timestamp: &Timestamp,
290-
command: &GenericCommandState,
290+
command: GenericCommandState,
291291
) -> Option<GenericCommandState> {
292-
let action = match self.get_action(command) {
292+
let action = match self.get_action(&command) {
293293
Ok(action) => action,
294294
Err(err) => {
295-
return Some(
296-
command
297-
.clone()
298-
.fail_with(format!("Fail to resume on start: {err:?}")),
299-
);
295+
return Some(command.fail_with(format!("Fail to resume on start: {err:?}")));
300296
}
301297
};
302298

303299
let epoch = format!("{}.{}", timestamp.unix_timestamp(), timestamp.millisecond());
304-
let command = command.clone().update_with_key_value("resumed_at", &epoch);
300+
let command = command.with_key_value("resumed_at", &epoch);
305301
match action {
306302
OperationAction::AwaitingAgentRestart(handlers) => {
307303
Some(command.update(handlers.on_success))

0 commit comments

Comments
 (0)