Added cron server event. (#349)

* Added cron server event.

The cron server event will be called whenever Redis runs its cron jobs (usually a few times per second).

Example:

```rust
#[cron_event_handler]
fn cron_event_handler(ctx: &Context, hz: u64) {
    // run some code here that should run periodically.
}
```

Author: Meir Shpilraien (Spielrein)

examples/server_events.rs

@@ -3,9 +3,10 @@ use std::sync::atomic::{AtomicI64, Ordering};
 use redis_module::{
     redis_module, server_events::FlushSubevent, Context, RedisResult, RedisString, RedisValue,
 };
-use redis_module_macros::{config_changed_event_handler, flush_event_handler};
+use redis_module_macros::{config_changed_event_handler, cron_event_handler, flush_event_handler};
 
 static NUM_FLUSHES: AtomicI64 = AtomicI64::new(0);
+static NUM_CRONS: AtomicI64 = AtomicI64::new(0);
 static NUM_MAX_MEMORY_CONFIGURATION_CHANGES: AtomicI64 = AtomicI64::new(0);
 
 #[flush_event_handler]
@@ -23,10 +24,19 @@ fn config_changed_event_handler(_ctx: &Context, changed_configs: &[&str]) {
         .map(|_| NUM_MAX_MEMORY_CONFIGURATION_CHANGES.fetch_add(1, Ordering::SeqCst));
 }
 
+#[cron_event_handler]
+fn cron_event_handler(_ctx: &Context, _hz: u64) {
+    NUM_CRONS.fetch_add(1, Ordering::SeqCst);
+}
+
 fn num_flushed(_ctx: &Context, _args: Vec<RedisString>) -> RedisResult {
     Ok(RedisValue::Integer(NUM_FLUSHES.load(Ordering::SeqCst)))
 }
 
+fn num_crons(_ctx: &Context, _args: Vec<RedisString>) -> RedisResult {
+    Ok(RedisValue::Integer(NUM_CRONS.load(Ordering::SeqCst)))
+}
+
 fn num_maxmemory_changes(_ctx: &Context, _args: Vec<RedisString>) -> RedisResult {
     Ok(RedisValue::Integer(
         NUM_MAX_MEMORY_CONFIGURATION_CHANGES.load(Ordering::SeqCst),
@@ -43,5 +53,6 @@ redis_module! {
     commands: [
         ["num_flushed", num_flushed, "readonly", 0, 0, 0],
         ["num_max_memory_changes", num_maxmemory_changes, "readonly", 0, 0, 0],
+        ["num_crons", num_crons, "readonly", 0, 0, 0],
     ],
 }

redismodule-rs-macros/src/lib.rs

@@ -195,6 +195,28 @@ pub fn config_changed_event_handler(_attr: TokenStream, item: TokenStream) -> To
     gen.into()
 }
 
+/// Proc macro which is set on a function that need to be called on Redis cron.
+/// The function must accept a [Context] and [u64] that represent the cron hz.
+///
+/// Example:
+///
+/// ```rust,no_run,ignore
+/// #[cron_event_handler]
+/// fn cron_event_handler(ctx: &Context, hz: u64) { ... }
+/// ```
+#[proc_macro_attribute]
+pub fn cron_event_handler(_attr: TokenStream, item: TokenStream) -> TokenStream {
+    let ast: ItemFn = match syn::parse(item) {
+        Ok(res) => res,
+        Err(e) => return e.to_compile_error().into(),
+    };
+    let gen = quote! {
+        #[linkme::distributed_slice(redis_module::server_events::CRON_SERVER_EVENTS_LIST)]
+        #ast
+    };
+    gen.into()
+}
+
 /// The macro auto generate a [From] implementation that can convert the struct into [RedisValue].
 ///
 /// Example:

src/context/call_reply.rs

@@ -109,6 +109,13 @@ pub enum ErrorReply<'root> {
     RedisError(ErrorCallReply<'root>),
 }
 
+/// Send implementation to [ErrorCallReply].
+/// We need to implements this trait because [ErrorCallReply] hold
+/// raw pointers to C data which does not auto implement the [Send] trait.
+/// By implementing [Send] on [ErrorCallReply] we basically tells the compiler
+/// that it is safe to send the underline C data between threads.
+unsafe impl<'root> Send for ErrorCallReply<'root> {}
+
 impl<'root> ErrorReply<'root> {
     /// Convert [ErrorCallReply] to [String] or [None] if its not a valid utf8.
     pub fn to_utf8_string(&self) -> Option<String> {
@@ -632,6 +639,13 @@ pub enum CallReply<'root> {
     VerbatimString(VerbatimStringCallReply<'root>),
 }
 
+/// Send implementation to [CallReply].
+/// We need to implements this trait because [CallReply] hold
+/// raw pointers to C data which does not auto implement the [Send] trait.
+/// By implementing [Send] on [CallReply] we basically tells the compiler
+/// that it is safe to send the underline C data between threads.
+unsafe impl<'root> Send for CallReply<'root> {}
+
 impl<'root> Display for CallReply<'root> {
     fn fmt(&self, f: &mut Formatter<'_>) -> fmt::Result {
         match self {

src/context/server_events.rs

@@ -54,6 +54,23 @@ pub static MODULE_CHANGED_SERVER_EVENTS_LIST: [fn(&Context, ModuleChangeSubevent
 #[distributed_slice()]
 pub static CONFIG_CHANGED_SERVER_EVENTS_LIST: [fn(&Context, &[&str])] = [..];
 
+#[distributed_slice()]
+pub static CRON_SERVER_EVENTS_LIST: [fn(&Context, u64)] = [..];
+
+extern "C" fn cron_callback(
+    ctx: *mut raw::RedisModuleCtx,
+    _eid: raw::RedisModuleEvent,
+    _subevent: u64,
+    data: *mut ::std::os::raw::c_void,
+) {
+    let data: &raw::RedisModuleConfigChangeV1 =
+        unsafe { &*(data as *mut raw::RedisModuleConfigChangeV1) };
+    let ctx = Context::new(ctx);
+    CRON_SERVER_EVENTS_LIST.iter().for_each(|callback| {
+        callback(&ctx, data.version);
+    });
+}
+
 extern "C" fn role_changed_callback(
     ctx: *mut raw::RedisModuleCtx,
     _eid: raw::RedisModuleEvent,
@@ -212,5 +229,11 @@ pub fn register_server_events(ctx: &Context) -> Result<(), RedisError> {
         raw::REDISMODULE_EVENT_CONFIG,
         Some(config_change_event_callback),
     )?;
+    register_single_server_event_type(
+        ctx,
+        &CRON_SERVER_EVENTS_LIST,
+        raw::REDISMODULE_EVENT_CRON_LOOP,
+        Some(cron_callback),
+    )?;
     Ok(())
 }

tests/integration.rs

@@ -411,6 +411,10 @@ fn test_server_event() -> Result<()> {
 
     assert_eq!(res, 2);
 
+    let res: i64 = redis::cmd("num_crons").query(&mut con)?;
+
+    assert!(res > 0);
+
     Ok(())
 }