Merge pull request #12 from CoLearn-Dev/init_func

- support init function

Author: stneng

Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "colink"
-version = "0.1.16"
+version = "0.1.17"
 edition = "2021"
 description = "CoLink Rust SDK"
 license = "MIT"

README.md

@@ -9,7 +9,7 @@ CoLink SDK helps both application adnd protocol developers access the functional
 Add this to your Cargo.toml:
 ```toml
 [dependencies]
-colink = "0.1.15"
+colink = "0.1.17"
 ```
 
 ## Getting Started

examples/protocol_greetings_with_init_func.rs

@@ -0,0 +1,52 @@
+#![allow(unused_variables)]
+use colink::{CoLink, Participant, ProtocolEntry};
+
+struct Init;
+#[colink::async_trait]
+impl ProtocolEntry for Init {
+    async fn start(
+        &self,
+        cl: CoLink,
+        _param: Vec<u8>,                 // For init function, param is empty
+        _participants: Vec<Participant>, // For init function, participants is empty
+    ) -> Result<(), Box<dyn std::error::Error + Send + Sync + 'static>> {
+        // Init
+        Ok(())
+    }
+}
+
+struct Initiator;
+#[colink::async_trait]
+impl ProtocolEntry for Initiator {
+    async fn start(
+        &self,
+        cl: CoLink,
+        param: Vec<u8>,
+        participants: Vec<Participant>,
+    ) -> Result<(), Box<dyn std::error::Error + Send + Sync + 'static>> {
+        println!("initiator");
+        Ok(())
+    }
+}
+
+struct Receiver;
+#[colink::async_trait]
+impl ProtocolEntry for Receiver {
+    async fn start(
+        &self,
+        cl: CoLink,
+        param: Vec<u8>,
+        participants: Vec<Participant>,
+    ) -> Result<(), Box<dyn std::error::Error + Send + Sync + 'static>> {
+        println!("{}", String::from_utf8_lossy(&param));
+        cl.create_entry(&format!("tasks:{}:output", cl.get_task_id()?), &param)
+            .await?;
+        Ok(())
+    }
+}
+
+colink::protocol_start!(
+    ("greetings:@init", Init),          // bind init function
+    ("greetings:initiator", Initiator), // bind initiator's entry function
+    ("greetings:receiver", Receiver)    // bind receiver's entry function
+);

src/protocol.rs

@@ -160,7 +160,54 @@ pub fn _protocol_start(
     cl: CoLink,
     user_funcs: HashMap<String, Box<dyn ProtocolEntry + Send + Sync>>,
 ) -> Result<(), Error> {
+    let mut operator_funcs: HashMap<String, Box<dyn ProtocolEntry + Send + Sync>> = HashMap::new();
+    let mut protocols = vec![];
     for (protocol_and_role, user_func) in user_funcs {
+        let cl = cl.clone();
+        if protocol_and_role.ends_with(":@init") {
+            let protocol_name = protocol_and_role[..protocol_and_role.len() - 6].to_string();
+            let _ = tokio::runtime::Builder::new_multi_thread()
+                .enable_all()
+                .build()
+                .unwrap()
+                .block_on(async move {
+                    let is_initialized_key =
+                        format!("_internal:protocols:{}:_is_initialized", protocol_name);
+                    let lock = cl.lock(&is_initialized_key).await?;
+                    let res = cl.read_entry(&is_initialized_key).await;
+                    if res.is_err() || res.unwrap()[0] == 0 {
+                        let cl_clone = cl.clone();
+                        match user_func
+                            .start(cl_clone, Default::default(), Default::default())
+                            .await
+                        {
+                            Ok(_) => {}
+                            Err(e) => error!("{}: {}.", protocol_and_role, e),
+                        }
+                        cl.update_entry(&is_initialized_key, &[1]).await?;
+                    }
+                    cl.unlock(lock).await?;
+                    Ok::<(), Box<dyn std::error::Error + Send + Sync + 'static>>(())
+                });
+        } else {
+            protocols.push(protocol_and_role[..protocol_and_role.rfind(':').unwrap()].to_string());
+            operator_funcs.insert(protocol_and_role, user_func);
+        }
+    }
+    let cl_clone = cl.clone();
+    let _ = tokio::runtime::Builder::new_multi_thread()
+        .enable_all()
+        .build()
+        .unwrap()
+        .block_on(async move {
+            for protocol_name in protocols {
+                let is_initialized_key =
+                    format!("_internal:protocols:{}:_is_initialized", protocol_name);
+                cl_clone.update_entry(&is_initialized_key, &[1]).await?;
+            }
+            Ok::<(), Box<dyn std::error::Error + Send + Sync + 'static>>(())
+        });
+    for (protocol_and_role, user_func) in operator_funcs {
         let cl = cl.clone();
         thread::spawn(|| {
             tokio::runtime::Builder::new_multi_thread()