feat(pubsub): implement core PubSub framework infrastructure (#103)

Signed-off-by: Joe Brinkman <[email protected]>
Signed-off-by: Joseph Brinkman <[email protected]>

Author: jbrinkman

.github/workflows/codeql.yml

@@ -34,7 +34,7 @@ jobs:
                   submodules: true
 
             - name: Initialize CodeQL
-              uses: github/codeql-action/init@v3
+              uses: github/codeql-action/init@v4
               with:
                   languages: csharp
                   build-mode: manual
@@ -50,6 +50,6 @@ jobs:
               run: dotnet build sources/Valkey.Glide/Valkey.Glide.csproj --configuration Lint --framework net8.0
 
             - name: Perform CodeQL Analysis
-              uses: github/codeql-action/analyze@v3
+              uses: github/codeql-action/analyze@v4
               with:
                   category: "/language:csharp"

.gitignore

@@ -31,8 +31,7 @@ x64/
 [Tt]est[Rr]esult*/
 [Bb]uild[Ll]og.*
 
-# Code Coverage Reports
-[Rr]eports/
+
 
 *_i.c
 *_p.c
@@ -156,8 +155,9 @@ _NCrunch*
 
 glide-logs/
 
-# Test results and reports
-reports/
+# Code Coverage Reports
+[Rr]eports/
+
 testresults/
 
 # Temporary submodules (not for commit)

rust/src/ffi.rs

@@ -7,7 +7,8 @@ use std::{
 
 use glide_core::{
     client::{
-        ConnectionRequest, ConnectionRetryStrategy, NodeAddress, ReadFrom as coreReadFrom, TlsMode,
+        AuthenticationInfo as CoreAuthenticationInfo, ConnectionRequest, ConnectionRetryStrategy,
+        NodeAddress, ReadFrom as coreReadFrom, TlsMode,
     },
     request_type::RequestType,
 };
@@ -71,16 +72,96 @@ pub struct ConnectionConfig {
     /// zero pointer is valid, means no client name is given (`None`)
     pub client_name: *const c_char,
     pub lazy_connect: bool,
+    pub refresh_topology_from_initial_nodes: bool,
+    pub has_pubsub_config: bool,
+    pub pubsub_config: PubSubConfigInfo,
     /*
     TODO below
     pub periodic_checks: Option<PeriodicCheck>,
-    pub pubsub_subscriptions: Option<redis::PubSubSubscriptionInfo>,
     pub inflight_requests_limit: Option<u32>,
     pub otel_endpoint: Option<String>,
     pub otel_flush_interval_ms: Option<u64>,
     */
 }
 
+#[repr(C)]
+#[derive(Debug, Clone, Copy)]
+pub struct PubSubConfigInfo {
+    pub channels_ptr: *const *const c_char,
+    pub channel_count: u32,
+    pub patterns_ptr: *const *const c_char,
+    pub pattern_count: u32,
+    pub sharded_channels_ptr: *const *const c_char,
+    pub sharded_channel_count: u32,
+}
+
+/// Convert a C string array to a Vec of Vec<u8>
+///
+/// # Safety
+///
+/// * `ptr` must point to an array of `count` valid C string pointers
+/// * Each C string pointer must be valid and null-terminated
+unsafe fn convert_string_array(ptr: *const *const c_char, count: u32) -> Vec<Vec<u8>> {
+    if ptr.is_null() || count == 0 {
+        return Vec::new();
+    }
+
+    let slice = unsafe { std::slice::from_raw_parts(ptr, count as usize) };
+    slice
+        .iter()
+        .map(|&str_ptr| {
+            let c_str = unsafe { CStr::from_ptr(str_ptr) };
+            c_str.to_bytes().to_vec()
+        })
+        .collect()
+}
+
+/// Convert PubSubConfigInfo to the format expected by glide-core
+///
+/// # Safety
+///
+/// * All pointers in `config` must be valid or null
+/// * String arrays must contain valid C strings
+unsafe fn convert_pubsub_config(
+    config: &PubSubConfigInfo,
+) -> std::collections::HashMap<redis::PubSubSubscriptionKind, std::collections::HashSet<Vec<u8>>> {
+    use redis::PubSubSubscriptionKind;
+    use std::collections::{HashMap, HashSet};
+
+    let mut subscriptions = HashMap::new();
+
+    // Convert exact channels
+    if config.channel_count > 0 {
+        let channels = unsafe { convert_string_array(config.channels_ptr, config.channel_count) };
+        subscriptions.insert(
+            PubSubSubscriptionKind::Exact,
+            channels.into_iter().collect::<HashSet<_>>(),
+        );
+    }
+
+    // Convert patterns
+    if config.pattern_count > 0 {
+        let patterns = unsafe { convert_string_array(config.patterns_ptr, config.pattern_count) };
+        subscriptions.insert(
+            PubSubSubscriptionKind::Pattern,
+            patterns.into_iter().collect::<HashSet<_>>(),
+        );
+    }
+
+    // Convert sharded channels
+    if config.sharded_channel_count > 0 {
+        let sharded = unsafe {
+            convert_string_array(config.sharded_channels_ptr, config.sharded_channel_count)
+        };
+        subscriptions.insert(
+            PubSubSubscriptionKind::Sharded,
+            sharded.into_iter().collect::<HashSet<_>>(),
+        );
+    }
+
+    subscriptions
+}
+
 /// Convert connection configuration to a corresponding object.
 ///
 /// # Safety
@@ -135,7 +216,7 @@ pub(crate) unsafe fn create_connection_request(
                 None
             };
 
-            Some(glide_core::client::AuthenticationInfo {
+            Some(CoreAuthenticationInfo {
                 username: unsafe { ptr_to_opt_str(auth_info.username) },
                 password: unsafe { ptr_to_opt_str(auth_info.password) },
                 iam_config,
@@ -172,10 +253,19 @@ pub(crate) unsafe fn create_connection_request(
             None
         },
         lazy_connect: config.lazy_connect,
-        refresh_topology_from_initial_nodes: false,
+        refresh_topology_from_initial_nodes: config.refresh_topology_from_initial_nodes,
+        pubsub_subscriptions: if config.has_pubsub_config {
+            let subscriptions = unsafe { convert_pubsub_config(&config.pubsub_config) };
+            if subscriptions.is_empty() {
+                None
+            } else {
+                Some(subscriptions)
+            }
+        } else {
+            None
+        },
         // TODO below
         periodic_checks: None,
-        pubsub_subscriptions: None,
         inflight_requests_limit: None,
     }
 }
@@ -636,3 +726,78 @@ pub(crate) unsafe fn get_pipeline_options(
         PipelineRetryStrategy::new(info.retry_server_error, info.retry_connection_error),
     )
 }
+
+/// FFI-safe version of [`redis::PushKind`] for C# interop.
+/// This enum maps to the `PushKind` enum in `sources/Valkey.Glide/Internals/FFI.structs.cs`.
+///
+/// The `#[repr(u32)]` attribute ensures a stable memory layout compatible with C# marshaling.
+/// Each variant corresponds to a specific Redis/Valkey PubSub notification type.
+#[repr(u32)]
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum PushKind {
+    /// Disconnection notification sent from the library when connection is closed.
+    Disconnection = 0,
+    /// Other/unknown push notification type.
+    Other = 1,
+    /// Cache invalidation notification received when a key is changed/deleted.
+    Invalidate = 2,
+    /// Regular channel message received via SUBSCRIBE.
+    Message = 3,
+    /// Pattern-based message received via PSUBSCRIBE.
+    PMessage = 4,
+    /// Sharded channel message received via SSUBSCRIBE.
+    SMessage = 5,
+    /// Unsubscribe confirmation.
+    Unsubscribe = 6,
+    /// Pattern unsubscribe confirmation.
+    PUnsubscribe = 7,
+    /// Sharded unsubscribe confirmation.
+    SUnsubscribe = 8,
+    /// Subscribe confirmation.
+    Subscribe = 9,
+    /// Pattern subscribe confirmation.
+    PSubscribe = 10,
+    /// Sharded subscribe confirmation.
+    SSubscribe = 11,
+}
+
+impl From<&redis::PushKind> for PushKind {
+    fn from(kind: &redis::PushKind) -> Self {
+        match kind {
+            redis::PushKind::Disconnection => PushKind::Disconnection,
+            redis::PushKind::Other(_) => PushKind::Other,
+            redis::PushKind::Invalidate => PushKind::Invalidate,
+            redis::PushKind::Message => PushKind::Message,
+            redis::PushKind::PMessage => PushKind::PMessage,
+            redis::PushKind::SMessage => PushKind::SMessage,
+            redis::PushKind::Unsubscribe => PushKind::Unsubscribe,
+            redis::PushKind::PUnsubscribe => PushKind::PUnsubscribe,
+            redis::PushKind::SUnsubscribe => PushKind::SUnsubscribe,
+            redis::PushKind::Subscribe => PushKind::Subscribe,
+            redis::PushKind::PSubscribe => PushKind::PSubscribe,
+            redis::PushKind::SSubscribe => PushKind::SSubscribe,
+        }
+    }
+}
+
+/// FFI callback function type for PubSub messages.
+/// This callback is invoked by Rust when a PubSub message is received.
+/// The callback signature matches the C# expectations for marshaling PubSub data.
+///
+/// # Parameters
+/// * `push_kind` - The type of push notification. See [`PushKind`] for valid values.
+/// * `message_ptr` - Pointer to the raw message bytes
+/// * `message_len` - Length of the message data in bytes (unsigned, cannot be negative)
+/// * `channel_ptr` - Pointer to the raw channel name bytes
+/// * `channel_len` - Length of the channel name in bytes (unsigned, cannot be negative)
+/// * `pattern_ptr` - Pointer to the raw pattern bytes (null if no pattern)
+/// * `pattern_len` - Length of the pattern in bytes (unsigned, 0 if no pattern)
+pub type PubSubCallback = unsafe extern "C" fn(
+    push_kind: PushKind,
+    message_ptr: *const u8,
+    message_len: u64,
+    channel_ptr: *const u8,
+    channel_len: u64,
+    pattern_ptr: *const u8,
+    pattern_len: u64,
+);