Skip to content

Commit 06d2c28

Browse files
stnengstneng
authored
Merge pull request #24 from CoLearn-Dev/issues-19-22
- wait_task example; initialization and other bugfix
2 parents 72e8f6f + 9ad5f32 commit 06d2c28

File tree

10 files changed

+90
-44
lines changed

10 files changed

+90
-44
lines changed

‎Cargo.toml‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,6 @@
11
[package]
22
name = "colink"
3-
version = "0.1.21"
3+
version = "0.1.22"
44
edition = "2021"
55
description = "CoLink Rust SDK"
66
license = "MIT"

‎README.md‎

Lines changed: 7 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -9,7 +9,7 @@ CoLink SDK helps both application adnd protocol developers access the functional
99
Add this to your Cargo.toml:
1010
```toml
1111
[dependencies]
12-
colink = "0.1.21"
12+
colink = "0.1.22"
1313
```
1414

1515
## Getting Started
@@ -38,6 +38,9 @@ cargo run --example user_run_task http://localhost:8080 $(sed -n "1,2p" ~/.colin
3838
cargo run --example host_import_user <address> <host_jwt> <expiration_timestamp> # <expiration_timestamp> is optional
3939
```
4040
```
41+
cargo run --example host_import_users <address> <host_jwt> <number> <expiration_timestamp> # <expiration_timestamp> is optional
42+
```
43+
```
4144
cargo run --example host_import_users_and_exchange_guest_jwts <address> <host_jwt> <number> <expiration_timestamp> # <expiration_timestamp> is optional
4245
```
4346
```
@@ -85,6 +88,9 @@ cargo run --example user_start_protocol_operator <address> <user_jwt> <protocol_
8588
```
8689
cargo run --example user_stop_protocol_operator <address> <user_jwt> <instance_id>
8790
```
91+
```
92+
cargo run --example user_wait_task <address> <user_jwt> <target_user_id>
93+
```
8894

8995
### Protocol
9096
```

‎examples/auto_confirm.rs‎

Lines changed: 6 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -1,4 +1,7 @@
1-
use colink::{CoLink, CoLinkInternalTaskIdList, StorageEntry, SubscriptionMessage, Task};
1+
use colink::{
2+
utils::get_path_timestamp, CoLink, CoLinkInternalTaskIdList, StorageEntry, SubscriptionMessage,
3+
Task,
4+
};
25
use prost::Message;
36
use std::env;
47
use tracing::debug;
@@ -37,11 +40,11 @@ async fn main() -> Result<(), Box<dyn std::error::Error + Send + Sync + 'static>
3740
let list_entry = &res[0];
3841
let list: CoLinkInternalTaskIdList = Message::decode(&*list_entry.payload).unwrap();
3942
if list.task_ids_with_key_paths.is_empty() {
40-
get_timestamp(&list_entry.key_path)
43+
get_path_timestamp(&list_entry.key_path)
4144
} else {
4245
list.task_ids_with_key_paths
4346
.iter()
44-
.map(|x| get_timestamp(&x.key_path))
47+
.map(|x| get_path_timestamp(&x.key_path))
4548
.min()
4649
.unwrap_or(i64::MAX)
4750
}
@@ -75,8 +78,3 @@ async fn main() -> Result<(), Box<dyn std::error::Error + Send + Sync + 'static>
7578
}
7679
}
7780
}
78-
79-
fn get_timestamp(key_path: &str) -> i64 {
80-
let pos = key_path.rfind('@').unwrap();
81-
key_path[pos + 1..].parse().unwrap()
82-
}

‎examples/user_wait_task.rs‎

Lines changed: 37 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,37 @@
1+
use colink::{
2+
decode_jwt_without_validation, extensions::registry::UserRecord, CoLink, Participant,
3+
};
4+
use prost::Message;
5+
use std::env;
6+
7+
#[tokio::main]
8+
async fn main() -> Result<(), Box<dyn std::error::Error + Send + Sync + 'static>> {
9+
let args = env::args().skip(1).collect::<Vec<_>>();
10+
let addr = &args[0];
11+
let jwt = &args[1];
12+
let target_user = &args[2];
13+
let user_id = decode_jwt_without_validation(jwt).unwrap().user_id;
14+
15+
let user = UserRecord {
16+
user_id: target_user.to_string(),
17+
..Default::default()
18+
};
19+
let mut payload = vec![];
20+
user.encode(&mut payload).unwrap();
21+
let cl = CoLink::new(addr, jwt);
22+
let participants = vec![Participant {
23+
user_id: user_id.to_string(),
24+
role: "query_from_registries".to_string(),
25+
}];
26+
let task_id = cl
27+
.run_task("registry", target_user.as_bytes(), &participants, false)
28+
.await?;
29+
println!(
30+
"Task {} has been created, waiting for it to finish...",
31+
task_id
32+
);
33+
cl.wait_task(&task_id).await?;
34+
println!("Task {} finished", task_id);
35+
36+
Ok(())
37+
}

‎src/extensions/policy_module.rs‎

Lines changed: 7 additions & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -1,4 +1,4 @@
1-
use crate::colink_proto::*;
1+
use crate::{colink_proto::*, utils::get_path_timestamp};
22
pub use colink_policy_module_proto::*;
33
use prost::Message;
44
mod colink_policy_module_proto {
@@ -19,7 +19,7 @@ impl crate::application::CoLink {
1919
{
2020
Ok(res) => (
2121
prost::Message::decode(&*res[0].payload)?,
22-
get_timestamp(&res[0].key_path),
22+
get_path_timestamp(&res[0].key_path),
2323
),
2424
Err(_) => (Default::default(), 0),
2525
};
@@ -30,7 +30,7 @@ impl crate::application::CoLink {
3030
settings.enable = true;
3131
let mut payload = vec![];
3232
settings.encode(&mut payload).unwrap();
33-
let timestamp = get_timestamp(
33+
let timestamp = get_path_timestamp(
3434
&self
3535
.update_entry("_policy_module:settings", &payload)
3636
.await?,
@@ -58,7 +58,7 @@ impl crate::application::CoLink {
5858
settings.enable = false;
5959
let mut payload = vec![];
6060
settings.encode(&mut payload).unwrap();
61-
let timestamp = get_timestamp(
61+
let timestamp = get_path_timestamp(
6262
&self
6363
.update_entry("_policy_module:settings", &payload)
6464
.await?,
@@ -88,7 +88,7 @@ impl crate::application::CoLink {
8888
settings.rules.push(rule);
8989
let mut payload = vec![];
9090
settings.encode(&mut payload).unwrap();
91-
let timestamp = get_timestamp(
91+
let timestamp = get_path_timestamp(
9292
&self
9393
.update_entry("_policy_module:settings", &payload)
9494
.await?,
@@ -109,7 +109,7 @@ impl crate::application::CoLink {
109109
settings.rules.retain(|x| x.rule_id != rule_id);
110110
let mut payload = vec![];
111111
settings.encode(&mut payload).unwrap();
112-
let timestamp = get_timestamp(
112+
let timestamp = get_path_timestamp(
113113
&self
114114
.update_entry("_policy_module:settings", &payload)
115115
.await?,
@@ -136,7 +136,7 @@ impl crate::application::CoLink {
136136
if applied_settings_timestamp >= timestamp {
137137
return Ok(());
138138
}
139-
get_timestamp(&res[0].key_path) + 1
139+
get_path_timestamp(&res[0].key_path) + 1
140140
}
141141
Err(_) => 0,
142142
};
@@ -157,8 +157,3 @@ impl crate::application::CoLink {
157157
Ok(())
158158
}
159159
}
160-
161-
fn get_timestamp(key_path: &str) -> i64 {
162-
let pos = key_path.rfind('@').unwrap();
163-
key_path[pos + 1..].parse().unwrap()
164-
}

‎src/extensions/registry.rs‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,5 @@
11
use crate::colink_proto::*;
2-
pub use colink_registry_proto::{Registries, Registry};
2+
pub use colink_registry_proto::{Registries, Registry, UserRecord};
33
use prost::Message;
44
mod colink_registry_proto {
55
include!(concat!(env!("OUT_DIR"), "/colink_registry.rs"));

‎src/extensions/wait_task.rs‎

Lines changed: 2 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -1,4 +1,4 @@
1-
use crate::colink_proto::*;
1+
use crate::{colink_proto::*, utils::get_path_timestamp};
22
use prost::Message;
33
use tracing::debug;
44

@@ -19,7 +19,7 @@ impl crate::application::CoLink {
1919
if task.status == "finished" {
2020
return Ok(());
2121
}
22-
get_timestamp(&res[0].key_path) + 1
22+
get_path_timestamp(&res[0].key_path) + 1
2323
}
2424
Err(_) => 0,
2525
};
@@ -40,8 +40,3 @@ impl crate::application::CoLink {
4040
Ok(())
4141
}
4242
}
43-
44-
fn get_timestamp(key_path: &str) -> i64 {
45-
let pos = key_path.rfind('@').unwrap();
46-
key_path[pos + 1..].parse().unwrap()
47-
}

‎src/lib.rs‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -12,3 +12,4 @@ pub use protocol::{
1212
CoLinkProtocol, ProtocolEntry, _colink_parse_args, _protocol_start, async_trait,
1313
};
1414
pub mod extensions;
15+
pub mod utils;

‎src/protocol.rs‎

Lines changed: 24 additions & 14 deletions
Original file line numberDiff line numberDiff line change
@@ -1,4 +1,4 @@
1-
use crate::application::*;
1+
use crate::{application::*, utils::get_path_timestamp};
22
pub use async_trait::async_trait;
33
use futures_lite::stream::StreamExt;
44
use lapin::{
@@ -7,7 +7,11 @@ use lapin::{
77
Connection, ConnectionProperties,
88
};
99
use prost::Message;
10-
use std::{collections::HashMap, sync::mpsc::channel, thread};
10+
use std::{
11+
collections::{HashMap, HashSet},
12+
sync::{mpsc::channel, Arc, Mutex},
13+
thread,
14+
};
1115
use structopt::StructOpt;
1216
use tracing::{debug, error};
1317

@@ -71,11 +75,11 @@ impl CoLinkProtocol {
7175
let list: CoLinkInternalTaskIdList =
7276
Message::decode(&*list_entry.payload).unwrap();
7377
if list.task_ids_with_key_paths.is_empty() {
74-
get_timestamp(&list_entry.key_path)
78+
get_path_timestamp(&list_entry.key_path)
7579
} else {
7680
list.task_ids_with_key_paths
7781
.iter()
78-
.map(|x| get_timestamp(&x.key_path))
82+
.map(|x| get_path_timestamp(&x.key_path))
7983
.min()
8084
.unwrap_or(i64::MAX)
8185
}
@@ -151,19 +155,16 @@ impl CoLinkProtocol {
151155
}
152156
}
153157

154-
fn get_timestamp(key_path: &str) -> i64 {
155-
let pos = key_path.rfind('@').unwrap();
156-
key_path[pos + 1..].parse().unwrap()
157-
}
158-
159158
pub fn _protocol_start(
160159
cl: CoLink,
161160
user_funcs: HashMap<String, Box<dyn ProtocolEntry + Send + Sync>>,
162161
) -> Result<(), Error> {
163162
let mut operator_funcs: HashMap<String, Box<dyn ProtocolEntry + Send + Sync>> = HashMap::new();
164-
let mut protocols = vec![];
163+
let mut protocols = HashSet::new();
164+
let failed_protocols = Arc::new(Mutex::new(HashSet::new()));
165165
for (protocol_and_role, user_func) in user_funcs {
166166
let cl = cl.clone();
167+
let failed_protocols = failed_protocols.clone();
167168
if protocol_and_role.ends_with(":@init") {
168169
let protocol_name = protocol_and_role[..protocol_and_role.len() - 6].to_string();
169170
tokio::runtime::Builder::new_multi_thread()
@@ -181,19 +182,27 @@ pub fn _protocol_start(
181182
.start(cl_clone, Default::default(), Default::default())
182183
.await
183184
{
184-
Ok(_) => {}
185-
Err(e) => error!("{}: {}.", protocol_and_role, e),
185+
Ok(_) => {
186+
cl.update_entry(&is_initialized_key, &[1]).await?;
187+
}
188+
Err(e) => {
189+
error!("{}: {}.", protocol_and_role, e);
190+
failed_protocols.lock().unwrap().insert(protocol_name);
191+
}
186192
}
187-
cl.update_entry(&is_initialized_key, &[1]).await?;
188193
}
189194
cl.unlock(lock).await?;
190195
Ok::<(), Box<dyn std::error::Error + Send + Sync + 'static>>(())
191196
})?;
192197
} else {
193-
protocols.push(protocol_and_role[..protocol_and_role.rfind(':').unwrap()].to_string());
198+
protocols
199+
.insert(protocol_and_role[..protocol_and_role.rfind(':').unwrap()].to_string());
194200
operator_funcs.insert(protocol_and_role, user_func);
195201
}
196202
}
203+
for failed_protocol in &*failed_protocols.lock().unwrap() {
204+
protocols.remove(failed_protocol);
205+
}
197206
let cl_clone = cl.clone();
198207
tokio::runtime::Builder::new_multi_thread()
199208
.enable_all()
@@ -259,6 +268,7 @@ pub struct CommandLineArgs {
259268
}
260269

261270
pub fn _colink_parse_args() -> CoLink {
271+
tracing_subscriber::fmt::init();
262272
let CommandLineArgs {
263273
addr,
264274
jwt,

‎src/utils.rs‎

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,4 @@
1+
pub fn get_path_timestamp(key_path: &str) -> i64 {
2+
let pos = key_path.rfind('@').unwrap();
3+
key_path[pos + 1..].parse().unwrap()
4+
}

0 commit comments

Comments
 (0)