Merge pull request #40 from CoLearn-Dev/is-issues

- instant server: error redirect
- instant server: check mq connection before starting
- remove async in InstantRegistry::new()

Author: stneng

Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "colink"
-version = "0.2.6"
+version = "0.2.7"
 edition = "2021"
 description = "CoLink Rust SDK"
 license = "MIT"
@@ -23,6 +23,7 @@ lapin = "2.1"
 prost = "0.10"
 rand = { version = "0.8.4", features = ["std_rng"] }
 rcgen = { version = "0.10", optional = true }
+reqwest = { version = "0.11", default-features = false, features = ["rustls-tls-native-roots"], optional = true }
 secp256k1 = { version = "0.25", features = ["rand-std"] }
 serde = { version = "1.0", features = ["derive"] }
 serde_json = "1.0"
@@ -45,5 +46,5 @@ remote_storage = ["extensions"]
 variable_transfer = ["extensions", "remote_storage", "hyper", "jsonwebtoken", "rcgen", "tokio-rustls", "hyper-rustls"]
 registry = []
 policy_module = []
-instant_server = []
+instant_server = ["reqwest"]
 storage_macro = ["async-recursion"]

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.2.5"
+colink = "0.2.7"
 ```
 
 ## Getting Started

examples/protocol_greetings_with_instant_server.rs

@@ -36,7 +36,7 @@ impl ProtocolEntry for Receiver {
 
 #[tokio::main]
 async fn main() -> Result<(), Box<dyn std::error::Error + Send + Sync + 'static>> {
-    let ir = InstantRegistry::new().await;
+    let ir = InstantRegistry::new();
     // The instant server will automatically use the instant registry when there is one.
     let is0 = InstantServer::new();
     let is1 = InstantServer::new();

src/extensions/instant_server.rs

@@ -70,6 +70,22 @@ impl InstantServer {
         } else {
             "http://guest:guest@localhost:15672/api".to_string()
         };
+        let (mq_amqp, mq_api) = std::thread::spawn(move || {
+            tokio::runtime::Builder::new_multi_thread()
+                .enable_all()
+                .build()
+                .unwrap()
+                .block_on(async {
+                    let res = reqwest::get(&mq_api).await.unwrap();
+                    assert!(res.status() == hyper::StatusCode::OK);
+                    lapin::Connection::connect(&mq_amqp, lapin::ConnectionProperties::default())
+                        .await
+                        .unwrap();
+                });
+            (mq_amqp, mq_api)
+        })
+        .join()
+        .unwrap();
         let child = Command::new(program)
             .args([
                 "--address",
@@ -88,8 +104,6 @@ impl InstantServer {
             ])
             .env("COLINK_HOME", colink_home)
             .current_dir(working_dir.clone())
-            .stdout(Stdio::null())
-            .stderr(Stdio::null())
             .spawn()
             .unwrap();
         loop {
@@ -129,17 +143,34 @@ impl Drop for InstantRegistry {
     }
 }
 
+impl Default for InstantRegistry {
+    fn default() -> Self {
+        Self::new()
+    }
+}
+
 impl InstantRegistry {
-    pub async fn new() -> Self {
+    pub fn new() -> Self {
         let is = InstantServer::new();
         let colink_home = get_colink_home().unwrap();
         let registry_file = Path::new(&colink_home).join("reg_config");
         let _file = File::options()
             .write(true)
             .create_new(true)
-            .open(&registry_file)
+            .open(registry_file)
             .unwrap();
-        is.get_colink().switch_to_generated_user().await.unwrap();
+        let is = std::thread::spawn(move || {
+            tokio::runtime::Builder::new_multi_thread()
+                .enable_all()
+                .build()
+                .unwrap()
+                .block_on(async {
+                    is.get_colink().switch_to_generated_user().await.unwrap();
+                });
+            is
+        })
+        .join()
+        .unwrap();
         Self {
             _instant_server: is,
         }

tests/test_protocol_variable_transfer.rs

@@ -55,7 +55,7 @@ impl ProtocolEntry for Receiver {
 
 #[tokio::test]
 async fn test_vt() -> Result<(), Box<dyn std::error::Error + Send + Sync + 'static>> {
-    let ir = InstantRegistry::new().await;
+    let ir = InstantRegistry::new();
     let mut iss = vec![];
     let mut cls = vec![];
     for i in 0..8 {