Effector stream (#142)

* basic implementation of effector stream on runtime-async-std, runtime-tokio

* resolve runtime-async-std warnings

* resolve runtime-tokio warnings

* fix casbin::error::Error

* bump version

Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "casbin"
-version = "0.7.6"
+version = "0.8.0"
 authors = ["Joey <joey.xf@gmail.com>", "Cheng JIANG <jiang.cheng@vip.163.com>"]
 edition = "2018"
 license = "Apache-2.0"
@@ -29,8 +29,8 @@ thiserror = "1.0.14"
 [features]
 default = ["runtime-async-std", "incremental"]
 
-runtime-tokio = ["tokio/fs", "tokio/io-util"]
-runtime-async-std = ["async-std"]
+runtime-tokio = ["tokio/fs", "tokio/io-util", "tokio/sync"]
+runtime-async-std = ["async-std/std", "async-std/unstable"]
 logging = ["log"]
 ip = ["ip_network"]
 glob = ["globset"]

README.md

@@ -27,7 +27,7 @@ Add this package to `Cargo.toml` of your project. (Check https://crates.io/crate
 
 ```toml
 [dependencies]
-casbin = { version = "0.7.6", default-features = false, features = ["runtime-async-std", "logging"] }
+casbin = { version = "0.8.0", default-features = false, features = ["runtime-async-std", "logging"] }
 async-std = { version = "1.5.0", features = ["attributes"] }
 env_logger = "0.7.1"
 ```

src/effector.rs

@@ -1,67 +1,90 @@
+use async_trait::async_trait;
+
+#[cfg(feature = "runtime-async-std")]
+use async_std::{sync::Receiver, task};
+
+#[cfg(feature = "runtime-tokio")]
+use tokio::{sync::mpsc::Receiver, task};
+
+#[async_trait]
 pub trait Effector: Send + Sync {
-    fn merge_effects(&self, expr: &str, effects: Vec<EffectKind>) -> bool;
+    #[allow(unused_mut)]
+    async fn merge_effects(&self, expr: &str, rx: Receiver<EffectKind>) -> bool;
+    fn clone_box(&self) -> Box<dyn Effector>;
 }
 
-#[derive(PartialEq, Clone)]
+#[derive(PartialEq, Clone, Debug)]
 pub enum EffectKind {
     Allow = 0,
     Indeterminate = 1,
     Deny = 2,
 }
 
-#[derive(Default)]
+#[derive(Default, Clone)]
 pub struct DefaultEffector;
 
+#[async_trait]
 impl Effector for DefaultEffector {
-    fn merge_effects(&self, expr: &str, effects: Vec<EffectKind>) -> bool {
-        if expr == "some(where (p_eft == allow))" {
-            let mut result = false;
-            for eft in effects {
-                if eft == EffectKind::Allow {
-                    result = true;
-                    break;
-                }
-            }
-
-            result
-        } else if expr == "!some(where (p_eft == deny))" {
-            let mut result = true;
-            for eft in effects {
-                if eft == EffectKind::Deny {
-                    result = false;
-                    break;
-                }
-            }
-
-            result
-        } else if expr == "some(where (p_eft == allow)) && !some(where (p_eft == deny))" {
-            let mut result = false;
-            for eft in effects {
-                if eft == EffectKind::Allow {
-                    result = true;
-                } else if eft == EffectKind::Deny {
-                    result = false;
-                    break;
-                }
-            }
+    #[allow(unused_mut)]
+    async fn merge_effects(&self, expr: &str, mut rx: Receiver<EffectKind>) -> bool {
+        let expr = expr.to_string();
+        let fut = task::spawn(async move {
+            let mut result = match &*expr {
+                "some(where (p_eft == allow))"
+                | "some(where (p_eft == allow)) && !some(where (p_eft == deny))"
+                | "priority(p_eft) || deny" => false,
+                "!some(where (p_eft == deny))" => true,
+                _ => panic!("unsupported effect: `{}`", expr),
+            };
 
-            result
-        } else if expr == "priority(p_eft) || deny" {
-            let mut result = false;
-            for eft in effects {
-                if eft != EffectKind::Indeterminate {
+            while let Some(eft) = rx.recv().await {
+                if &expr == "some(where (p_eft == allow))" {
+                    if eft == EffectKind::Allow {
+                        result = true;
+                        break;
+                    }
+                } else if &expr == "!some(where (p_eft == deny))" {
+                    if eft == EffectKind::Deny {
+                        result = false;
+                        break;
+                    }
+                } else if &expr == "some(where (p_eft == allow)) && !some(where (p_eft == deny))" {
                     if eft == EffectKind::Allow {
-                        result = true
-                    } else {
-                        result = false
+                        result = true;
+                    } else if eft == EffectKind::Deny {
+                        result = false;
+                        break;
+                    }
+                } else if &expr == "priority(p_eft) || deny" {
+                    if eft != EffectKind::Indeterminate {
+                        if eft == EffectKind::Allow {
+                            result = true
+                        } else {
+                            result = false
+                        }
+                        break;
                     }
-                    break;
                 }
             }
 
             result
-        } else {
-            panic!("unsupported effect: `{}`", expr);
+        });
+
+        #[cfg(feature = "runtime-async-std")]
+        {
+            fut.await
+        }
+
+        #[cfg(feature = "runtime-tokio")]
+        {
+            match fut.await {
+                Ok(result) => result,
+                Err(err) => panic!("effector stream error: {}", err),
+            }
         }
     }
+
+    fn clone_box(&self) -> Box<dyn Effector> {
+        Box::new(self.clone())
+    }
 }

src/enforcer.rs

@@ -21,6 +21,12 @@ use crate::watcher::Watcher;
 #[cfg(feature = "logging")]
 use crate::{DefaultLogger, Logger};
 
+#[cfg(feature = "runtime-async-std")]
+use async_std::sync::channel;
+
+#[cfg(feature = "runtime-tokio")]
+use tokio::sync::mpsc::channel;
+
 use async_trait::async_trait;
 use rhai::{
     def_package,
@@ -144,13 +150,14 @@ impl Enforcer {
             }
         }
 
-        let mut policy_effects: Vec<EffectKind> = vec![];
         let policies = p_ast.get_policy();
-        let policy_len = policies.len();
+        let policies_len = policies.len();
         let scope_size = scope.len();
 
-        if policy_len != 0 {
-            policy_effects = vec![EffectKind::Deny; policy_len];
+        #[allow(unused_mut)]
+        let (mut tx, mut rx) = channel(if policies_len > 0 { policies_len } else { 1 });
+
+        if policies_len != 0 {
             for (i, pvals) in policies.iter().enumerate() {
                 if i != 0 {
                     scope.rewind(scope_size);
@@ -176,33 +183,35 @@ impl Enforcer {
                     .engine
                     .eval_with_scope::<bool>(&mut scope, &expstring)?;
                 if !eval_result {
-                    policy_effects[i] = EffectKind::Indeterminate;
+                    #[cfg(feature = "runtime-async-std")]
+                    {
+                        tx.send(EffectKind::Indeterminate).await;
+                    }
+                    #[cfg(feature = "runtime-tokio")]
+                    {
+                        tx.send(EffectKind::Indeterminate).await?;
+                    }
                     continue;
                 }
-                if let Some(j) = p_ast.tokens.iter().position(|x| x == "p_eft") {
+                let eft = if let Some(j) = p_ast.tokens.iter().position(|x| x == "p_eft") {
                     let eft = &pvals[j];
                     if eft == "allow" {
-                        policy_effects[i] = EffectKind::Allow;
+                        EffectKind::Allow
                     } else if eft == "deny" {
-                        policy_effects[i] = EffectKind::Deny;
+                        EffectKind::Deny
                     } else {
-                        policy_effects[i] = EffectKind::Indeterminate;
+                        EffectKind::Indeterminate
                     }
                 } else {
-                    policy_effects[i] = EffectKind::Allow;
-                }
-                if e_ast.value == "priority(p_eft) || deny" {
-                    break;
-                } else if e_ast.value == "some(where (p_eft == allow))"
-                    && policy_effects[i] == EffectKind::Allow
+                    EffectKind::Allow
+                };
+                #[cfg(feature = "runtime-async-std")]
                 {
-                    return Ok(true);
-                } else if policy_effects[i] == EffectKind::Deny
-                    && (e_ast.value == "!some(where (p_eft == deny))"
-                        || e_ast.value
-                            == "some(where (p_eft == allow)) && !some(where (p_eft == deny))")
+                    tx.send(eft).await;
+                }
+                #[cfg(feature = "runtime-tokio")]
                 {
-                    return Ok(false);
+                    tx.send(eft).await?;
                 }
             }
         } else {
@@ -212,14 +221,25 @@ impl Enforcer {
             let eval_result = self
                 .engine
                 .eval_with_scope::<bool>(&mut scope, &m_ast.value)?;
-            if eval_result {
-                policy_effects.push(EffectKind::Allow);
+            let eft = if eval_result {
+                EffectKind::Allow
             } else {
-                policy_effects.push(EffectKind::Indeterminate);
+                EffectKind::Deny
+            };
+            #[cfg(feature = "runtime-async-std")]
+            {
+                tx.send(eft).await;
+            }
+            #[cfg(feature = "runtime-tokio")]
+            {
+                tx.send(eft).await?;
             }
         }
 
-        Ok(self.eft.merge_effects(&e_ast.value, policy_effects))
+        let effector = self.eft.clone_box();
+        std::mem::drop(tx);
+
+        Ok(effector.merge_effects(&e_ast.value, rx).await)
     }
 }
 

src/error.rs

@@ -1,6 +1,12 @@
+#[cfg(feature = "runtime-tokio")]
+use crate::effector::EffectKind;
+
 use rhai::EvalAltResult;
 use thiserror::Error;
 
+#[cfg(feature = "runtime-tokio")]
+use tokio::sync::mpsc::error::SendError;
+
 use std::{error::Error as StdError, io::Error as IoError};
 
 /// ModelError represents any type of errors in model configuration
@@ -66,6 +72,10 @@ pub enum Error {
 
     #[error("Casbin Adapter Error: `{0:?}`")]
     AdapterError(#[from] AdapterError),
+
+    #[cfg(feature = "runtime-tokio")]
+    #[error("Tokio Channel Error: `{0:?}`")]
+    ChannelError(#[from] SendError<EffectKind>),
 }
 
 #[cfg(test)]