fix, query onlines, block current ffi thread

Signed-off-by: dignow <linlong1265@gmail.com>

src/flutter.rs

@@ -1597,3 +1597,71 @@ pub mod sessions {
             .unwrap_or(false)
     }
 }
+
+pub(super) mod async_tasks {
+    use hbb_common::{
+        bail,
+        tokio::{
+            select,
+            sync::mpsc::{unbounded_channel, UnboundedSender},
+        },
+        ResultType,
+    };
+    use std::{
+        collections::HashMap,
+        sync::{Arc, Mutex},
+    };
+
+    type TxQueryOnlines = UnboundedSender<Vec<String>>;
+    lazy_static::lazy_static! {
+        static ref TX_QUERY_ONLINES: Arc<Mutex<Option<TxQueryOnlines>>> = Default::default();
+    }
+
+    pub fn start_flutter_async_runner() {
+        std::thread::spawn(|| async {
+            let (tx_onlines, mut rx_onlines) = unbounded_channel::<Vec<String>>();
+            TX_QUERY_ONLINES.lock().unwrap().replace(tx_onlines);
+            loop {
+                select! {
+                    ids = rx_onlines.recv() => {
+                        match ids {
+                            Some(_ids) => {
+                                #[cfg(not(any(target_os = "ios")))]
+                                crate::rendezvous_mediator::query_online_states(_ids, handle_query_onlines)
+                            }
+                            None => {
+                                break;
+                            }
+                        }
+                    }
+                }
+            }
+        });
+    }
+
+    #[allow(dead_code)]
+    pub fn stop_flutter_async_runner() {
+        let _ = TX_QUERY_ONLINES.lock().unwrap().take();
+    }
+
+    pub fn query_onlines(ids: Vec<String>) -> ResultType<()> {
+        if let Some(tx) = TX_QUERY_ONLINES.lock().unwrap().as_ref() {
+            tx.send(ids)?;
+        } else {
+            bail!("No tx_query_onlines");
+        }
+        Ok(())
+    }
+
+    fn handle_query_onlines(onlines: Vec<String>, offlines: Vec<String>) {
+        let data = HashMap::from([
+            ("name", "callback_query_onlines".to_owned()),
+            ("onlines", onlines.join(",")),
+            ("offlines", offlines.join(",")),
+        ]);
+        let _res = super::push_global_event(
+            super::APP_TYPE_MAIN,
+            serde_json::ser::to_string(&data).unwrap_or("".to_owned()),
+        );
+    }
+}

src/flutter_ffi.rs

@@ -36,6 +36,7 @@ lazy_static::lazy_static! {
 }
 
 fn initialize(app_dir: &str) {
+    flutter::async_tasks::start_flutter_async_runner();
     *config::APP_DIR.write().unwrap() = app_dir.to_owned();
     #[cfg(target_os = "android")]
     {
@@ -1554,18 +1555,6 @@ pub fn main_get_build_date() -> String {
     crate::BUILD_DATE.to_string()
 }
 
-fn handle_query_onlines(onlines: Vec<String>, offlines: Vec<String>) {
-    let data = HashMap::from([
-        ("name", "callback_query_onlines".to_owned()),
-        ("onlines", onlines.join(",")),
-        ("offlines", offlines.join(",")),
-    ]);
-    let _res = flutter::push_global_event(
-        flutter::APP_TYPE_MAIN,
-        serde_json::ser::to_string(&data).unwrap_or("".to_owned()),
-    );
-}
-
 pub fn translate(name: String, locale: String) -> SyncReturn<String> {
     SyncReturn(crate::client::translate_locale(name, &locale))
 }
@@ -1589,8 +1578,7 @@ pub fn session_register_texture(
 }
 
 pub fn query_onlines(ids: Vec<String>) {
-    #[cfg(not(any(target_os = "ios")))]
-    crate::rendezvous_mediator::query_online_states(ids, handle_query_onlines)
+    let _ = flutter::async_tasks::query_onlines(ids);
 }
 
 pub fn version_to_number(v: String) -> SyncReturn<i64> {

src/rendezvous_mediator.rs

@@ -598,7 +598,7 @@ pub async fn query_online_states<F: FnOnce(Vec<String>, Vec<String>)>(ids: Vec<S
             }
 
             if query_begin.elapsed() > query_timeout {
-                log::debug!("query onlines timeout {:?}", query_timeout);
+                log::debug!("query onlines timeout {:?} ({:?})", query_begin.elapsed(), query_timeout);
                 break;
             }