Added API to register to server events.

.github/workflows/cratesio-publish.yml

@@ -28,7 +28,22 @@ jobs:
           key: "package.version"
           value: "${{ steps.get_version.outputs.VERSION }}"
 
-      - uses: katyo/publish-crates@v1
+      - name: Set the version for publishing on derive crate
+        uses: ciiiii/toml-editor@1.0.0
+        with:
+          file: "redismodule-rs-derive/Cargo.toml"
+          key: "package.version"
+          value: "${{ steps.get_version.outputs.VERSION }}"
+
+      - name: Publishing redismodule-rs
+        uses: katyo/publish-crates@v1
+        with:
+          registry-token: ${{ secrets.CARGO_REGISTRY_TOKEN }}
+          args: --allow-dirty
+
+      - name: Publishing redismodule-rs-derive
+        uses: katyo/publish-crates@v1
         with:
           registry-token: ${{ secrets.CARGO_REGISTRY_TOKEN }}
+          path: './redismodule-rs-derive'
           args: --allow-dirty

Cargo.toml

@@ -58,6 +58,10 @@ crate-type = ["cdylib"]
 name = "ctx_flags"
 crate-type = ["cdylib"]
 
+[[example]]
+name = "server_events"
+crate-type = ["cdylib"]
+
 [[example]]
 name = "events"
 crate-type = ["cdylib"]
@@ -90,10 +94,12 @@ regex = "1"
 strum_macros = "0.24"
 #failure = "0.1"
 backtrace = "0.3"
+linkme = "0.3"
 
 [dev-dependencies]
 anyhow = "1.0.38"
 redis = "0.22.1"
+redis-module-derive = { path = "./redismodule-rs-derive"}
 
 [build-dependencies]
 bindgen = "0.64"

examples/server_events.rs

@@ -0,0 +1,31 @@
+#[macro_use]
+extern crate redis_module;
+
+use std::sync::atomic::{AtomicI64, Ordering};
+
+use redis_module::{server_events::FlushSubevent, Context, RedisResult, RedisString, RedisValue};
+use redis_module_derive::flush_event_handler;
+
+static NUM_FLUSHES: AtomicI64 = AtomicI64::new(0);
+
+#[flush_event_handler]
+fn flushed_event_handler(_ctx: &Context, flush_event: FlushSubevent) {
+    if let FlushSubevent::Started = flush_event {
+        NUM_FLUSHES.fetch_add(1, Ordering::SeqCst);
+    }
+}
+
+fn num_flushed(_ctx: &Context, _args: Vec<RedisString>) -> RedisResult {
+    Ok(RedisValue::Integer(NUM_FLUSHES.load(Ordering::SeqCst)))
+}
+
+//////////////////////////////////////////////////////
+
+redis_module! {
+    name: "server_events",
+    version: 1,
+    data_types: [],
+    commands: [
+        ["num_flushed", num_flushed, "readonly", 0, 0, 0],
+    ],
+}

redismodule-rs-derive/Cargo.toml

@@ -0,0 +1,21 @@
+[package]
+name = "redis-module-derive"
+version = "99.99.99"
+authors = ["Meir Shpilraien <meir@redis.com>"]
+edition = "2021"
+description = "A derive crate for redismodule-rs"
+license = "BSD-3-Clause"
+repository = "https://github.com/RedisLabsModules/redismodule-rs"
+keywords = ["redis", "plugin"]
+categories = ["database", "api-bindings"]
+
+# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
+
+[dependencies]
+syn = { version="1.0", features = ["full"]}
+quote = "1.0"
+
+[lib]
+name = "redis_module_derive"
+path = "src/lib.rs"
+proc-macro = true

redismodule-rs-derive/src/lib.rs

@@ -0,0 +1,57 @@
+extern crate proc_macro;
+use proc_macro::TokenStream;
+use quote::quote;
+use syn;
+use syn::ItemFn;
+
+#[proc_macro_attribute]
+pub fn role_changed_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::ROLE_CHANGED_SERVER_EVENTS_LIST)]
+        #ast
+    };
+    gen.into()
+}
+
+#[proc_macro_attribute]
+pub fn loading_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::LOADING_SERVER_EVENTS_LIST)]
+        #ast
+    };
+    gen.into()
+}
+
+#[proc_macro_attribute]
+pub fn flush_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::FLUSH_SERVER_EVENTS_LIST)]
+        #ast
+    };
+    gen.into()
+}
+
+#[proc_macro_attribute]
+pub fn module_changed_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::MODULE_CHANGED_SERVER_EVENTS_LIST)]
+        #ast
+    };
+    gen.into()
+}

src/context/mod.rs

@@ -23,6 +23,8 @@ pub mod blocked;
 
 pub mod info;
 
+pub mod server_events;
+
 pub mod keys_cursor;
 
 /// `Context` is a structure that's designed to give us a high-level interface to

src/context/server_events.rs

@@ -0,0 +1,194 @@
+use crate::raw;
+use crate::{context::Context, RedisError};
+use linkme::distributed_slice;
+
+#[derive(Clone)]
+pub enum ServerRole {
+    Primary,
+    Replica,
+}
+
+#[derive(Clone)]
+pub enum LoadingSubevent {
+    RdbStarted,
+    AofStarted,
+    ReplStarted,
+    Ended,
+    Failed,
+}
+
+#[derive(Clone)]
+pub enum FlushSubevent {
+    Started,
+    Ended,
+}
+
+#[derive(Clone)]
+pub enum ModuleChangeSubevent {
+    Loaded,
+    Unloaded,
+}
+
+pub enum ServerEventHandler {
+    RuleChanged(fn(&Context, ServerRole)),
+    Loading(fn(&Context, LoadingSubevent)),
+    Flush(fn(&Context, FlushSubevent)),
+    ModuleChange(fn(&Context, ModuleChangeSubevent)),
+}
+
+#[distributed_slice()]
+pub static ROLE_CHANGED_SERVER_EVENTS_LIST: [fn(&Context, ServerRole)] = [..];
+
+#[distributed_slice()]
+pub static LOADING_SERVER_EVENTS_LIST: [fn(&Context, LoadingSubevent)] = [..];
+
+#[distributed_slice()]
+pub static FLUSH_SERVER_EVENTS_LIST: [fn(&Context, FlushSubevent)] = [..];
+
+#[distributed_slice()]
+pub static MODULE_CHANGED_SERVER_EVENTS_LIST: [fn(&Context, ModuleChangeSubevent)] = [..];
+
+extern "C" fn role_changed_callback(
+    ctx: *mut raw::RedisModuleCtx,
+    _eid: raw::RedisModuleEvent,
+    subevent: u64,
+    _data: *mut ::std::os::raw::c_void,
+) {
+    let new_role = if subevent == raw::REDISMODULE_EVENT_REPLROLECHANGED_NOW_MASTER {
+        ServerRole::Primary
+    } else {
+        ServerRole::Replica
+    };
+    let ctx = Context::new(ctx);
+    ROLE_CHANGED_SERVER_EVENTS_LIST.iter().for_each(|callback| {
+        callback(&ctx, new_role.clone());
+    });
+}
+
+extern "C" fn loading_event_callback(
+    ctx: *mut raw::RedisModuleCtx,
+    _eid: raw::RedisModuleEvent,
+    subevent: u64,
+    _data: *mut ::std::os::raw::c_void,
+) {
+    let loading_sub_event = match subevent {
+        raw::REDISMODULE_SUBEVENT_LOADING_RDB_START => LoadingSubevent::RdbStarted,
+        raw::REDISMODULE_SUBEVENT_LOADING_REPL_START => LoadingSubevent::ReplStarted,
+        raw::REDISMODULE_SUBEVENT_LOADING_ENDED => LoadingSubevent::Ended,
+        _ => LoadingSubevent::Failed,
+    };
+    let ctx = Context::new(ctx);
+    LOADING_SERVER_EVENTS_LIST.iter().for_each(|callback| {
+        callback(&ctx, loading_sub_event.clone());
+    });
+}
+
+extern "C" fn flush_event_callback(
+    ctx: *mut raw::RedisModuleCtx,
+    _eid: raw::RedisModuleEvent,
+    subevent: u64,
+    _data: *mut ::std::os::raw::c_void,
+) {
+    let flush_sub_event = if subevent == raw::REDISMODULE_SUBEVENT_FLUSHDB_START {
+        FlushSubevent::Started
+    } else {
+        FlushSubevent::Ended
+    };
+    let ctx = Context::new(ctx);
+    FLUSH_SERVER_EVENTS_LIST.iter().for_each(|callback| {
+        callback(&ctx, flush_sub_event.clone());
+    });
+}
+
+extern "C" fn module_change_event_callback(
+    ctx: *mut raw::RedisModuleCtx,
+    _eid: raw::RedisModuleEvent,
+    subevent: u64,
+    _data: *mut ::std::os::raw::c_void,
+) {
+    let module_changed_sub_event = if subevent == raw::REDISMODULE_SUBEVENT_MODULE_LOADED {
+        ModuleChangeSubevent::Loaded
+    } else {
+        ModuleChangeSubevent::Unloaded
+    };
+    let ctx = Context::new(ctx);
+    MODULE_CHANGED_SERVER_EVENTS_LIST
+        .iter()
+        .for_each(|callback| {
+            callback(&ctx, module_changed_sub_event.clone());
+        });
+}
+
+pub fn register_server_events(ctx: &Context) -> Result<(), RedisError> {
+    if !ROLE_CHANGED_SERVER_EVENTS_LIST.is_empty() {
+        let res = unsafe {
+            raw::RedisModule_SubscribeToServerEvent.unwrap()(
+                ctx.ctx,
+                raw::RedisModuleEvent {
+                    id: raw::REDISMODULE_EVENT_REPLICATION_ROLE_CHANGED,
+                    dataver: 1,
+                },
+                Some(role_changed_callback),
+            )
+        };
+        if res != raw::REDISMODULE_OK as i32 {
+            return Err(RedisError::Str(
+                "Failed subscribing to role changed server event",
+            ));
+        }
+    }
+
+    if !LOADING_SERVER_EVENTS_LIST.is_empty() {
+        let res = unsafe {
+            raw::RedisModule_SubscribeToServerEvent.unwrap()(
+                ctx.ctx,
+                raw::RedisModuleEvent {
+                    id: raw::REDISMODULE_EVENT_LOADING,
+                    dataver: 1,
+                },
+                Some(loading_event_callback),
+            )
+        };
+        if res != raw::REDISMODULE_OK as i32 {
+            return Err(RedisError::Str(
+                "Failed subscribing to loading server event",
+            ));
+        }
+    }
+
+    if !FLUSH_SERVER_EVENTS_LIST.is_empty() {
+        let res = unsafe {
+            raw::RedisModule_SubscribeToServerEvent.unwrap()(
+                ctx.ctx,
+                raw::RedisModuleEvent {
+                    id: raw::REDISMODULE_EVENT_FLUSHDB,
+                    dataver: 1,
+                },
+                Some(flush_event_callback),
+            )
+        };
+        if res != raw::REDISMODULE_OK as i32 {
+            return Err(RedisError::Str("Failed subscribing to flush server event"));
+        }
+    }
+
+    if !MODULE_CHANGED_SERVER_EVENTS_LIST.is_empty() {
+        let res = unsafe {
+            raw::RedisModule_SubscribeToServerEvent.unwrap()(
+                ctx.ctx,
+                raw::RedisModuleEvent {
+                    id: raw::REDISMODULE_EVENT_MODULE_CHANGE,
+                    dataver: 1,
+                },
+                Some(module_change_event_callback),
+            )
+        };
+        if res != raw::REDISMODULE_OK as i32 {
+            return Err(RedisError::Str(
+                "Failed subscribing to module changed server event",
+            ));
+        }
+    }
+
+    Ok(())
+}

src/lib.rs

@@ -28,6 +28,7 @@ pub use crate::context::thread_safe::{DetachedFromClient, ThreadSafeContext};
 pub use crate::raw::NotifyEvent;
 
 pub use crate::context::keys_cursor::KeysCursor;
+pub use crate::context::server_events;
 pub use crate::context::Context;
 pub use crate::context::ContextFlags;
 pub use crate::raw::*;

src/macros.rs

@@ -132,6 +132,7 @@ macro_rules! redis_module {
 
             use $crate::raw;
             use $crate::RedisString;
+            use $crate::server_events::register_server_events;
 
             // We use a statically sized buffer to avoid allocating.
             // This is needed since we use a custom allocator that relies on the Redis allocator,
@@ -181,6 +182,11 @@ macro_rules! redis_module {
 
             raw::register_info_function(ctx, Some(__info_func));
 
+            if let Err(e) = register_server_events(&context) {
+                context.log_warning(&format!("{e}"));
+                return raw::Status::Err as c_int;
+            }
+
             raw::Status::Ok as c_int
         }