create a fake persitenceUpstream example

- expose the struct Wrappers for as they will be required to build custom peristence upstreams

Cargo.toml

@@ -11,7 +11,7 @@ description = "A basic implementation of GCRA algorithm for rate limiting"
 keywords = ["rate-limit", "rate", "limit", "gcra"]
 
 [features]
-default = []
+default = ["rate-limiter"]
 rate-limiter = ["thingvellir", "tokio"]
 
 [dependencies]

examples/rate_limiter.rs

@@ -1,9 +1,12 @@
 use gcra::{GcraError, RateLimit, RateLimiter, RateLimiterError};
 
+const CACHE_CAPACITY: usize = 4;
+const WORKER_SHARD_COUNT: u8 = 4;
+
 #[tokio::main]
 async fn main() -> Result<(), RateLimiterError> {
     let rate_limit = RateLimit::per_sec(2);
-    let mut rl = RateLimiter::new(4, 4);
+    let mut rl = RateLimiter::new(CACHE_CAPACITY, WORKER_SHARD_COUNT);
 
     rl.check("key", rate_limit.clone(), 1).await?;
     rl.check("key", rate_limit.clone(), 1).await?;

examples/rate_limiter_persisted.rs

@@ -0,0 +1,86 @@
+use std::{collections::HashMap, time::Duration, hash::Hash, thread, sync::{RwLock, Arc}};
+
+use gcra::{GcraError, RateLimit, RateLimiter, RateLimiterError, RateLimitEntry};
+use thingvellir::{LoadFromUpstream, CommitToUpstream};
+
+
+const CACHE_CAPACITY: usize = 4;
+const WORKER_SHARD_COUNT: u8 = 4;
+
+const IO_DURATION: Duration = Duration::from_millis(50);
+
+#[tokio::main]
+async fn main() -> Result<(), RateLimiterError> {
+    let rate_limit = RateLimit::per_sec(2);
+
+    // Use a persistence backed rate limiter
+    let upstream_factory = FakePersistenceUpstream {
+        data: Arc::new(RwLock::new(HashMap::new()))
+    };
+    let mut rl = RateLimiter::with_handle(
+        thingvellir::service_builder(CACHE_CAPACITY)
+            .num_shards(WORKER_SHARD_COUNT)
+            .build_mutable(upstream_factory, thingvellir::DefaultCommitPolicy::Immediate)
+    );
+
+    rl.check("key", rate_limit.clone(), 1).await?;
+    rl.check("key", rate_limit.clone(), 1).await?;
+
+    match rl.check("key", rate_limit.clone(), 1).await {
+        Err(RateLimiterError::GcraError(GcraError::DeniedUntil { next_allowed_at })) => {
+            println!("Denied: Request next at {:?}", next_allowed_at);
+
+            let duration = Duration::from_millis(10);
+            println!("Sleep for {:?} to allow commits", duration);
+            tokio::time::sleep(duration).await;
+        }
+        unexpected => panic!("Opps something went wrong! {:?}", unexpected),
+    };
+
+    Ok(())
+}
+
+#[derive(Clone)]
+struct FakePersistenceUpstream<K, D>{data: Arc<RwLock<HashMap<K, D>>>}
+
+
+impl<Key> LoadFromUpstream<Key, RateLimitEntry> for FakePersistenceUpstream<Key, RateLimitEntry>
+where
+    Key: 'static + Send + Sync + Hash + Eq + Clone,
+{
+    fn load(&mut self, request: thingvellir::DataLoadRequest<Key, RateLimitEntry>) {
+        let key = request.key().clone();
+        let data = self.data.clone();
+
+        request.spawn_default(async move {
+            println!("LOAD. Sleeping for {:?}", IO_DURATION);
+            tokio::time::sleep(IO_DURATION).await;
+
+            match data.read().expect("RwLock poisoned").get(&key) {
+                Some(value) => {
+                    println!("LOADED {:?}", value);
+                    Ok(value.clone())},
+                None => {
+                    println!("LOADED NOT_FOUND");
+                    Err(thingvellir::UpstreamError::KeyNotFound)
+                },
+            }
+        });
+    }
+}
+
+impl<Key> CommitToUpstream<Key, RateLimitEntry> for FakePersistenceUpstream<Key, RateLimitEntry>
+where 
+    Key: 'static + Send + Sync + Hash + Eq + Clone,
+{
+    fn commit<'a>(&mut self, request: thingvellir::DataCommitRequest<'a, Key, RateLimitEntry>) {
+        let key = request.key().clone();
+        let data = request.data().clone();
+
+        println!("COMMIT. Sleeping for {:?}", IO_DURATION);
+        thread::sleep(IO_DURATION);
+
+        self.data.write().expect("RwLock poisoned").insert(key, data);
+        request.resolve()
+    }
+}

src/lib.rs

@@ -55,4 +55,4 @@ mod rate_limiter;
 pub use crate::gcra::{GcraError, GcraState};
 pub use crate::rate_limit::RateLimit;
 #[cfg(feature = "rate-limiter")]
-pub use crate::rate_limiter::{RateLimiter, RateLimiterError};
+pub use crate::rate_limiter::{RateLimiter, RateLimiterError, RateLimitEntry, RateLimitRequest};

src/rate_limiter/entry.rs

@@ -9,8 +9,8 @@ use thingvellir::{
 
 use crate::{GcraState, RateLimit};
 
-#[derive(Default, Clone)]
-pub(crate) struct RateLimitEntry {
+#[derive(Default, Debug, Clone)]
+pub struct RateLimitEntry {
     pub gcra_state: GcraState,
     expires_at: Option<tokio::time::Instant>,
 }
@@ -31,7 +31,6 @@ impl DerefMut for RateLimitEntry {
 
 impl ServiceData for RateLimitEntry {
     fn should_persist(&self) -> bool {
-        // we are in memory, always save to
         true
     }
 

src/rate_limiter/mod.rs

@@ -3,3 +3,4 @@ mod entry;
 mod rate_limiter;
 
 pub use rate_limiter::*;
+pub use entry::*;

src/rate_limiter/rate_limiter.rs

@@ -14,7 +14,7 @@ pub enum RateLimiterError {
 }
 
 #[derive(Clone, Hash, PartialEq, Eq, Debug)]
-struct RateLimitRequest<T> {
+pub struct RateLimitRequest<T> {
     key: T,
 }
 
@@ -39,6 +39,7 @@ impl<Key> RateLimiter<Key, InstantClock>
 where
     Key: Send + Clone + Hash + Eq + Display + 'static,
 {
+    /// Constructs an [InMemoryUpstream] sharded instance of a rate limiter.
     pub fn new(max_data_capacity: usize, num_shards: u8) -> Self {
         Self {
             clock: InstantClock,
@@ -50,6 +51,14 @@ where
                 ),
         }
     }
+
+    /// Constructs a instance using the [MutableServiceHandle] provided.
+    pub fn with_handle(shard_handle: MutableServiceHandle<RateLimitRequest<Key>, RateLimitEntry>) -> Self {
+        Self {
+            clock: InstantClock,
+            shard_handle
+        }
+    }
 }
 
 impl<Key, C> RateLimiter<Key, C>
@@ -92,7 +101,7 @@ where
                 match check_result {
                     Ok(_) => {
                         entry.update_expiration(&rate_limit);
-                        Commit::immediately(check_result)
+                        Commit::default(check_result)
                     }
                     Err(GcraError::DeniedUntil { .. })
                     | Err(GcraError::DeniedIndefinitely { .. }) => unsafe {