Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

feat(spells): store last trigger by trigger key in KV [NET-511] #1728

Merged
merged 3 commits into from
Jul 31, 2023
Merged

feat(spells): store last trigger by trigger key in KV [NET-511] #1728

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
7001411
feat(spells): store last trigger by `trigger` key in KV
justprosh Jul 28, 2023
473238d
update tests
justprosh Jul 28, 2023
3b9c05b
Merge branch 'master' into trigger-arg
justprosh Jul 31, 2023
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
56 changes: 16 additions & 40 deletions crates/nox-tests/tests/spells.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -901,12 +901,9 @@ async fn spell_timer_trigger_mailbox_test() {
(seq
(seq
(call %init_peer_id% ("getDataSrv" "spell_id") [] spell_id)
(call %init_peer_id% (spell_id "pop_mailbox") [] result)
)
(seq
(call %init_peer_id% ("json" "parse") [result.$.message.[0].message] obj)
(call "{}" ("return" "") [obj])
(call %init_peer_id% ("getDataSrv" "trigger") [] trigger)
)
(call "{}" ("return" "") [trigger])
)
"#,
client.peer_id
Expand All @@ -931,7 +928,7 @@ async fn spell_timer_trigger_mailbox_test() {
}

#[tokio::test]
async fn spell_connection_pool_trigger_mailbox_test() {
async fn spell_connection_pool_trigger_test() {
let swarms = make_swarms(1).await;
let mut client = ConnectedClient::connect_to(swarms[0].multiaddr.clone())
.await
Expand All @@ -945,14 +942,11 @@ async fn spell_connection_pool_trigger_mailbox_test() {
(seq
(call %init_peer_id% ("getDataSrv" "spell_id") [] spell_id)
(seq
(call %init_peer_id% (spell_id "pop_mailbox") [] trigger)
(call %init_peer_id% ("run-console" "print") ["pop mailbox, trigger:" trigger])
(call %init_peer_id% ("getDataSrv" "trigger") [] trigger)
(call %init_peer_id% ("run-console" "print") ["getDataSrv, trigger:" trigger])
)
)
(seq
(call %init_peer_id% ("json" "parse") [trigger.$.message.[0].message] obj)
(call "{}" ("return" "") [obj])
)
(call "{}" ("return" "") [trigger])
)
(call %init_peer_id% ("run-console" "print") ["herror" %last_error%])
)
Expand Down Expand Up @@ -1102,9 +1096,9 @@ async fn spell_update_config() {
r#"(seq
(seq
(call %init_peer_id% ("getDataSrv" "spell_id") [] spell_id)
(call %init_peer_id% (spell_id "pop_mailbox") [] result)
(call %init_peer_id% ("getDataSrv" "trigger") [] trigger)
)
(call "{}" ("return" "") [result])
(call "{}" ("return" "") [trigger])
)"#,
client.peer_id
);
Expand All @@ -1115,20 +1109,14 @@ async fn spell_update_config() {
.await
.unwrap();

if let [JValue::Object(x)] = client
if let [trigger] = client
.receive_args()
.await
.wrap_err("receive")
.unwrap()
.as_slice()
{
assert_eq!(x["absent"], JValue::Bool(false), "spell must be triggered");
let message = x["message"].as_array().unwrap()[0]
.as_object()
.cloned()
.unwrap();
let info: TriggerInfoAqua =
serde_json::from_str(message["message"].as_str().unwrap()).unwrap();
let info: TriggerInfoAqua = serde_json::from_str(&trigger.to_string()).unwrap();
let info: TriggerInfo = info.into();
assert_matches!(info, TriggerInfo::Peer(p) if p.connected, "spell must be triggered by the `connected` event");
} else {
Expand Down Expand Up @@ -1171,20 +1159,14 @@ async fn spell_update_config() {

drop(connected);

if let [JValue::Object(x)] = client
if let [trigger] = client
.receive_args()
.await
.wrap_err("receive")
.unwrap()
.as_slice()
{
assert_eq!(x["absent"], JValue::Bool(false), "spell must be triggered");
let message = x["message"].as_array().unwrap()[0]
.as_object()
.cloned()
.unwrap();
let info: TriggerInfoAqua =
serde_json::from_str(message["message"].as_str().unwrap()).unwrap();
let info: TriggerInfoAqua = serde_json::from_str(&trigger.to_string()).unwrap();
let info: TriggerInfo = info.into();
assert_matches!(info, TriggerInfo::Peer(p) if !p.connected, "spell must be triggered by the `disconnected` event");
} else {
Expand All @@ -1204,9 +1186,9 @@ async fn spell_update_config_stopped_spell() {
r#"(seq
(seq
(call %init_peer_id% ("getDataSrv" "spell_id") [] spell_id)
(call %init_peer_id% (spell_id "pop_mailbox") [] result)
(call %init_peer_id% ("getDataSrv" "trigger") [] trigger)
)
(call "{}" ("return" "") [result])
(call "{}" ("return" "") [trigger])
)"#,
client.peer_id
);
Expand Down Expand Up @@ -1248,20 +1230,14 @@ async fn spell_update_config_stopped_spell() {
};
assert_eq!(result, "done", "spell must be updated");

if let [JValue::Object(x)] = client
if let [trigger] = client
.receive_args()
.await
.wrap_err("receive")
.unwrap()
.as_slice()
{
assert_eq!(x["absent"], JValue::Bool(false), "spell must be triggered");
let message = x["message"].as_array().unwrap()[0]
.as_object()
.cloned()
.unwrap();
let info: TriggerInfoAqua =
serde_json::from_str(message["message"].as_str().unwrap()).unwrap();
let info: TriggerInfoAqua = serde_json::from_str(&trigger.to_string()).unwrap();
let info: TriggerInfo = info.into();
assert_matches!(
info,
Expand Down
42 changes: 13 additions & 29 deletions sorcerer/src/script_executor.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -127,38 +127,22 @@ impl Sorcerer {
event: TriggerEvent,
worker_id: PeerId,
) -> Result<(), JError> {
log::info!("storing trigger {:?}", event);
justprosh marked this conversation as resolved.
Show resolved Hide resolved
let serialized_event = serde_json::to_string(&TriggerInfoAqua::from(event.info))?;

let func_outcome = self.services.call_function(
worker_id,
process_func_outcome::<UnitValue>(
self.services.call_function(
worker_id,
&event.spell_id,
"set_string",
vec![json!("trigger"), json!(serialized_event)],
None,
worker_id,
self.spell_script_particle_ttl,
),
&event.spell_id,
"push_mailbox",
vec![json!(serialized_event)],
None,
worker_id,
self.spell_script_particle_ttl,
);

match process_func_outcome::<UnitValue>(func_outcome, &event.spell_id, "push_mailbox") {
Ok(_) => Ok(()),
Err(err) => {
log::warn!("Error on push_mailbox for spell {}: {}. Trying a fallback with list_push_string", event.spell_id, err);

// fallback for older spell versions
let func_outcome = self.services.call_function(
worker_id,
&event.spell_id,
"list_push_string",
vec![json!("trigger_mailbox"), json!(serialized_event)],
None,
worker_id,
self.spell_script_particle_ttl,
);
process_func_outcome::<UnitValue>(func_outcome, &event.spell_id, "list_push_string")
.map(drop)
}
}
"set_string",
)
.map(drop)
}

pub async fn execute_script(&self, event: TriggerEvent) {
Expand Down
Loading