API extentions.

Cargo.toml

@@ -61,6 +61,21 @@ name = "info"
 crate-type = ["cdylib"]
 required-features = []
 
+[[example]]
+name = "stream"
+crate-type = ["cdylib"]
+required-features = ["experimental-api"]
+
+[[example]]
+name = "server_events"
+crate-type = ["cdylib"]
+required-features = ["experimental-api"]
+
+[[example]]
+name = "scan"
+crate-type = ["cdylib"]
+required-features = ["experimental-api"]
+
 [dependencies]
 bitflags = "1.2"
 libc = "0.2"
@@ -70,6 +85,7 @@ regex = "1"
 strum_macros = "0.24"
 #failure = "0.1"
 backtrace = "0.3"
+nix = "0.26"
 
 [dev-dependencies]
 anyhow = "1.0.38"
@@ -82,8 +98,11 @@ cc = "1.0"
 [features]
 default = []
 experimental-api = []
-
-# Workaround to allow cfg(feature = "test") in dependencies:
-# https://github.com/rust-lang/rust/issues/59168#issuecomment-472653680
-# This requires running the tests with `--features test`
-test = []
+# Allows to fallback to the default Rust allocator when the Redis
+# allocator is not available. Particularly useful for tests without
+# having a running Redis instance and running as a Redis module.
+fallback_to_system_allocator = []
+# TODO remove this "feature" as its name is misleading and using of
+# which should be avoided in favour of well-named and self-explanatory
+# features doing just one job following the SRP and KISS principles.
+test = ["fallback_to_system_allocator"]

examples/data_type.rs

@@ -33,11 +33,16 @@ static MY_REDIS_TYPE: RedisType = RedisType::new(
         unlink: None,
         copy: None,
         defrag: None,
+
+        free_effort2: None,
+        unlink2: None,
+        copy2: None,
+        mem_usage2: None,
     },
 );
 
 unsafe extern "C" fn free(value: *mut c_void) {
-    Box::from_raw(value.cast::<MyType>());
+    drop(Box::from_raw(value.cast::<MyType>()));
 }
 
 fn alloc_set(ctx: &Context, args: Vec<RedisString>) -> RedisResult {

examples/events.rs

@@ -3,15 +3,17 @@ extern crate redis_module;
 
 use redis_module::{Context, NotifyEvent, RedisError, RedisResult, RedisString, Status};
 
-fn on_event(ctx: &Context, event_type: NotifyEvent, event: &str, key: &str) {
+fn on_event(ctx: &Context, event_type: NotifyEvent, event: &str, key: &[u8]) {
     let msg = format!(
         "Received event: {:?} on key: {} via event: {}",
-        event_type, key, event
+        event_type,
+        std::str::from_utf8(key).unwrap(),
+        event
     );
     ctx.log_debug(msg.as_str());
 }
 
-fn on_stream(ctx: &Context, _event_type: NotifyEvent, _event: &str, _key: &str) {
+fn on_stream(ctx: &Context, _event_type: NotifyEvent, _event: &str, _key: &[u8]) {
     ctx.log_debug("Stream event received!");
 }
 

examples/scan.rs

@@ -0,0 +1,28 @@
+#[macro_use]
+extern crate redis_module;
+
+use redis_module::{
+    context::keys_cursor::KeysCursor, Context, RedisResult, RedisString, RedisValue,
+};
+
+fn scan_keys(ctx: &Context, _args: Vec<RedisString>) -> RedisResult {
+    let mut keys = Vec::new();
+    let cursor = KeysCursor::new();
+    while cursor.scan(ctx, &|_ctx, key_name, _key| {
+        keys.push(RedisValue::BulkString(
+            key_name.try_as_str().unwrap().to_string(),
+        ));
+    }) {}
+    Ok(keys.into())
+}
+
+//////////////////////////////////////////////////////
+
+redis_module! {
+    name: "scan",
+    version: 1,
+    data_types: [],
+    commands: [
+        ["SCAN_KEYS", scan_keys, "fast deny-oom readonly", 0, 0, 0],
+    ],
+}

examples/server_events.rs

@@ -0,0 +1,55 @@
+#[macro_use]
+extern crate redis_module;
+
+use redis_module::{
+    context::server_events::ServerEventData, Context, RedisResult, RedisString, RedisValue,
+};
+
+static mut NUM_FLUSHES: usize = 0;
+static mut NUM_ROLED_CHANGED: usize = 0;
+static mut NUM_LOADINGS: usize = 0;
+
+fn num_flushed(_ctx: &Context, _args: Vec<RedisString>) -> RedisResult {
+    Ok(RedisValue::Integer(unsafe { NUM_FLUSHES } as i64))
+}
+
+fn num_roled_changed(_ctx: &Context, _args: Vec<RedisString>) -> RedisResult {
+    Ok(RedisValue::Integer(unsafe { NUM_ROLED_CHANGED } as i64))
+}
+
+fn num_loading(_ctx: &Context, _args: Vec<RedisString>) -> RedisResult {
+    Ok(RedisValue::Integer(unsafe { NUM_LOADINGS } as i64))
+}
+
+fn on_role_changed(_ctx: &Context, _event_data: ServerEventData) {
+    let num_roled_changed = unsafe { &mut NUM_ROLED_CHANGED };
+    *num_roled_changed = *num_roled_changed + 1;
+}
+
+fn on_loading_event(_ctx: &Context, _event_data: ServerEventData) {
+    let num_loading = unsafe { &mut NUM_LOADINGS };
+    *num_loading = *num_loading + 1;
+}
+
+fn on_flush_event(_ctx: &Context, _event_data: ServerEventData) {
+    let num_flushed = unsafe { &mut NUM_FLUSHES };
+    *num_flushed = *num_flushed + 1;
+}
+
+//////////////////////////////////////////////////////
+
+redis_module! {
+    name: "server_events",
+    version: 1,
+    data_types: [],
+    commands: [
+        ["NUM_FLUSHED", num_flushed, "fast deny-oom readonly", 0, 0, 0],
+        ["NUM_ROLED_CHANGED", num_roled_changed, "fast deny-oom readonly", 0, 0, 0],
+        ["NUM_LOADING", num_loading, "fast deny-oom readonly", 0, 0, 0],
+    ],
+    server_events: [
+        [@RuleChanged: on_role_changed],
+        [@Loading: on_loading_event],
+        [@Flush: on_flush_event],
+    ]
+}

examples/stream.rs

@@ -0,0 +1,48 @@
+#[macro_use]
+extern crate redis_module;
+
+use redis_module::raw::{KeyType, RedisModuleStreamID};
+use redis_module::{Context, NextArg, RedisError, RedisResult, RedisString, RedisValue};
+
+fn stream_read_from(ctx: &Context, args: Vec<RedisString>) -> RedisResult {
+    let mut args = args.into_iter().skip(1);
+
+    let stream_key = args.next_arg()?;
+
+    let stream = ctx.open_key(&stream_key);
+
+    let key_type = stream.key_type();
+
+    if key_type != KeyType::Stream {
+        return Err(RedisError::WrongType);
+    }
+
+    let mut iter = stream.get_stream_iterator()?;
+    let element = iter.next();
+    let id_to_keep = iter.next().as_ref().map_or_else(
+        || RedisModuleStreamID {
+            ms: u64::MAX,
+            seq: u64::MAX,
+        },
+        |e| e.id,
+    );
+
+    let stream = ctx.open_key_writable(&stream_key);
+    stream.trim_stream_by_id(id_to_keep, false)?;
+
+    Ok(match element {
+        Some(e) => RedisValue::BulkString(format!("{}-{}", e.id.ms, e.id.seq)),
+        None => RedisValue::Null,
+    })
+}
+
+//////////////////////////////////////////////////////
+
+redis_module! {
+    name: "stream",
+    version: 1,
+    data_types: [],
+    commands: [
+        ["STREAM_POP", stream_read_from, "fast deny-oom", 1, 1, 1],
+    ],
+}

examples/test_helper.rs

@@ -12,7 +12,6 @@ fn test_helper_version(ctx: &Context, _args: Vec<RedisString>) -> RedisResult {
     Ok(response.into())
 }
 
-#[cfg(feature = "test")]
 fn test_helper_version_rm_call(ctx: &Context, _args: Vec<RedisString>) -> RedisResult {
     let ver = ctx.get_redis_version_rm_call()?;
     let response: Vec<i64> = vec![ver.major.into(), ver.minor.into(), ver.patch.into()];
@@ -43,7 +42,6 @@ fn add_info(ctx: &InfoContext, _for_crash_report: bool) {
 
 //////////////////////////////////////////////////////
 
-#[cfg(feature = "test")]
 redis_module! {
     name: "test_helper",
     version: 1,

src/alloc.rs

@@ -1,29 +1,60 @@
 use std::alloc::{GlobalAlloc, Layout};
-use std::os::raw::c_void;
 
 use crate::raw;
 
+/// Panics with a message without using an allocator.
+/// Useful when using the allocator should be avoided or it is
+/// inaccessible. The default [std::panic] performs allocations and so
+/// will cause a double panic without a meaningful message if the
+/// allocator can't be used. This function makes sure we can panic with
+/// a reasonable message even without the allocator working.
+fn allocation_free_panic(message: &'static str) -> ! {
+    use std::os::unix::io::AsRawFd;
+
+    let _ = nix::unistd::write(std::io::stderr().as_raw_fd(), message.as_bytes());
+
+    std::process::abort();
+}
+
+const REDIS_ALLOCATOR_NOT_AVAILABLE_MESSAGE: &str =
+    "Critical error: the Redis Allocator isn't available.
+Consider enabling the \"fallback_to_system_allocator\" feature.\n";
+
+/// Defines the Redis allocator. This allocator delegates the allocation
+/// and deallocation tasks to the Redis server when available, otherwise
+/// it fallbacks to the default Rust [std::alloc::System] allocator
+/// which is always available compared to the Redis allocator.
+#[derive(Copy, Clone)]
 pub struct RedisAlloc;
 
 unsafe impl GlobalAlloc for RedisAlloc {
     unsafe fn alloc(&self, layout: Layout) -> *mut u8 {
-        /*
-         * To make sure the memory allocation by Redis is aligned to the according to the layout,
-         * we need to align the size of the allocation to the layout.
-         *
-         * "Memory is conceptually broken into equal-sized chunks,
-         * where the chunk size is a power of two that is greater than the page size.
-         * Chunks are always aligned to multiples of the chunk size.
-         * This alignment makes it possible to find metadata for user objects very quickly."
-         *
-         * From: https://linux.die.net/man/3/jemalloc
-         */
-        let size = (layout.size() + layout.align() - 1) & (!(layout.align() - 1));
-
-        raw::RedisModule_Alloc.unwrap()(size).cast::<u8>()
+        match raw::RedisModule_Alloc {
+            Some(alloc) => {
+                /*
+                 * To make sure the memory allocation by Redis is aligned to the
+                 * according to the layout, we need to align the size of the
+                 * allocation to the layout.
+                 *
+                 * "Memory is conceptually broken into equal-sized chunks,
+                 * where the chunk size is a power of two that is greater than
+                 * the page size. Chunks are always aligned to multiples of the
+                 * chunk size. This alignment makes it possible to find metadata
+                 * for user objects very quickly."
+                 *
+                 * From: https://linux.die.net/man/3/jemalloc
+                 */
+                let size = (layout.size() + layout.align() - 1) & (!(layout.align() - 1));
+                alloc(size).cast()
+            }
+            None => allocation_free_panic(REDIS_ALLOCATOR_NOT_AVAILABLE_MESSAGE),
+        }
     }
 
     unsafe fn dealloc(&self, ptr: *mut u8, _layout: Layout) {
-        raw::RedisModule_Free.unwrap()(ptr.cast::<c_void>())
+        match raw::RedisModule_Free {
+            Some(dealloc) => dealloc(ptr.cast()),
+            None => allocation_free_panic(REDIS_ALLOCATOR_NOT_AVAILABLE_MESSAGE),
+        }
     }
 }