finish rust integration

timkpaine · timkpaine · commit 61ead66291fe · 2026-02-24T17:17:39.000-05:00
Signed-off-by: Tim Paine &lt;3105306+timkpaine@users.noreply.github.com&gt;
diff --git a/examples/05_cpp/5_c_api_adapter_rust/src/bindings.rs b/examples/05_cpp/5_c_api_adapter_rust/src/bindings.rs
@@ -158,11 +158,53 @@ pub struct CCspAdapterManagerVTable {
 // We use weak linking / dynamic_lookup on macOS.
 // ============================================================================
 
-// Function pointer types for CSP C API
-pub type FnCcspEngineNow = unsafe extern "C" fn(engine: CCspEngineHandle) -> CCspDateTime;
-pub type FnCcspInputGetType = unsafe extern "C" fn(input: CCspInputHandle) -> CCspType;
-pub type FnCcspInputGetLastInt64 = unsafe extern "C" fn(input: CCspInputHandle, out: *mut i64) -> CCspErrorCode;
-pub type FnCcspPushInt64 = unsafe extern "C" fn(adapter: CCspPushInputAdapterHandle, value: i64, batch: *mut c_void) -> CCspErrorCode;
+// ============================================================================
+// CSP C API function declarations
+//
+// These are resolved at runtime via dynamic lookup (see build.rs).
+// The symbols are provided by CSP's shared library when the Python module
+// is loaded alongside CSP.
+// ============================================================================
+
+extern "C" {
+    // Engine functions
+    pub fn ccsp_engine_now(engine: CCspEngineHandle) -> CCspDateTime;
+
+    // Input functions (for output adapters)
+    pub fn ccsp_input_get_type(input: CCspInputHandle) -> CCspType;
+    pub fn ccsp_input_get_last_bool(input: CCspInputHandle, out: *mut i8) -> CCspErrorCode;
+    pub fn ccsp_input_get_last_int64(input: CCspInputHandle, out: *mut i64) -> CCspErrorCode;
+    pub fn ccsp_input_get_last_double(input: CCspInputHandle, out: *mut f64) -> CCspErrorCode;
+    pub fn ccsp_input_get_last_string(
+        input: CCspInputHandle,
+        out_data: *mut *const c_char,
+        out_length: *mut usize,
+    ) -> CCspErrorCode;
+    pub fn ccsp_input_get_last_datetime(input: CCspInputHandle, out: *mut CCspDateTime) -> CCspErrorCode;
+
+    // Push functions (for input adapters)
+    pub fn ccsp_push_input_adapter_push_bool(
+        adapter: CCspPushInputAdapterHandle,
+        value: i8,
+        batch: *mut c_void,
+    ) -> CCspErrorCode;
+    pub fn ccsp_push_input_adapter_push_int64(
+        adapter: CCspPushInputAdapterHandle,
+        value: i64,
+        batch: *mut c_void,
+    ) -> CCspErrorCode;
+    pub fn ccsp_push_input_adapter_push_double(
+        adapter: CCspPushInputAdapterHandle,
+        value: f64,
+        batch: *mut c_void,
+    ) -> CCspErrorCode;
+    pub fn ccsp_push_input_adapter_push_string(
+        adapter: CCspPushInputAdapterHandle,
+        data: *const c_char,
+        length: usize,
+        batch: *mut c_void,
+    ) -> CCspErrorCode;
+}
 
 // ============================================================================
 // Capsule creation helpers
diff --git a/examples/05_cpp/5_c_api_adapter_rust/src/input_adapter.rs b/examples/05_cpp/5_c_api_adapter_rust/src/input_adapter.rs
@@ -3,21 +3,24 @@
 //! This module demonstrates how to implement a CSP push input adapter in Rust
 //! using the C API FFI bindings.
 //!
-//! Note: The actual data pushing requires the CSP C API symbols to be available
-//! at runtime. When built standalone, the callbacks are implemented but the
-//! push functionality is stubbed.
+//! The adapter spawns a background thread that periodically pushes integer
+//! values to the CSP graph using the C API.
 
 use std::ffi::c_void;
-use std::sync::atomic::{AtomicBool, AtomicI64, Ordering};
+use std::ptr;
+use std::sync::atomic::{AtomicBool, AtomicI64, AtomicPtr, Ordering};
 use std::sync::Arc;
+use std::thread::{self, JoinHandle};
+use std::time::Duration;
 
 use crate::bindings::{
-    CCspDateTime, CCspEngineHandle, CCspPushInputAdapterHandle,
+    ccsp_push_input_adapter_push_int64, CCspDateTime, CCspEngineHandle,
+    CCspPushInputAdapterHandle,
 };
 
 /// Push input adapter that generates incrementing counter values.
 ///
-/// This adapter would spawn a background thread when started that periodically
+/// This adapter spawns a background thread when started that periodically
 /// pushes integer values to the CSP graph using the C API.
 pub struct RustInputAdapter {
     /// Interval between pushes in milliseconds
@@ -29,9 +32,11 @@ pub struct RustInputAdapter {
     /// Flag to signal thread to stop
     running: Arc<AtomicBool>,
 
-    /// Push adapter handle (set by start callback)
-    #[allow(dead_code)]
-    adapter_handle: Option<CCspPushInputAdapterHandle>,
+    /// Push adapter handle (stored as AtomicPtr for thread-safe access)
+    adapter_handle: Arc<AtomicPtr<c_void>>,
+
+    /// Thread handle
+    thread_handle: Option<JoinHandle<()>>,
 }
 
 impl RustInputAdapter {
@@ -41,36 +46,83 @@ impl RustInputAdapter {
             interval_ms: if interval_ms > 0 { interval_ms } else { 100 },
             counter: Arc::new(AtomicI64::new(0)),
             running: Arc::new(AtomicBool::new(false)),
-            adapter_handle: None,
+            adapter_handle: Arc::new(AtomicPtr::new(ptr::null_mut())),
+            thread_handle: None,
         }
     }
 
     /// Start the adapter.
     ///
-    /// In a full implementation with CSP C API symbols linked, this would
-    /// spawn a background thread that calls ccsp_push_input_adapter_push_int64.
+    /// Spawns a background thread that pushes integer values at the configured
+    /// interval using the CSP C API.
     pub fn start(&mut self, adapter: CCspPushInputAdapterHandle) {
-        self.adapter_handle = Some(adapter);
+        // Store the adapter handle (AtomicPtr is Send + Sync)
+        self.adapter_handle.store(adapter, Ordering::SeqCst);
         self.running.store(true, Ordering::SeqCst);
 
+        let running = Arc::clone(&self.running);
+        let counter = Arc::clone(&self.counter);
+        let adapter_handle = Arc::clone(&self.adapter_handle);
+        let interval_ms = self.interval_ms;
+
         eprintln!(
-            "[RustInputAdapter] Started with interval {} ms (adapter handle: {:?})",
+            "[RustInputAdapter] Starting with interval {} ms (adapter handle: {:?})",
             self.interval_ms, adapter
         );
 
-        // Note: In a full implementation, we would spawn a thread here that
-        // calls ccsp_push_input_adapter_push_int64 to push values.
-        // This requires the CSP C API symbols to be available at runtime.
+        // Spawn background thread that pushes values
+        let handle = thread::spawn(move || {
+            let interval = Duration::from_millis(interval_ms);
+
+            while running.load(Ordering::SeqCst) {
+                let value = counter.fetch_add(1, Ordering::SeqCst);
+                let adapter_ptr = adapter_handle.load(Ordering::SeqCst);
+
+                // Push the counter value to CSP using the C API
+                if !adapter_ptr.is_null() {
+                    unsafe {
+                        let result = ccsp_push_input_adapter_push_int64(
+                            adapter_ptr,
+                            value,
+                            ptr::null_mut(),
+                        );
+                        // Log any errors (in production, handle more gracefully)
+                        if result != crate::bindings::CCspErrorCode::Ok {
+                            eprintln!(
+                                "[RustInputAdapter] Push failed with error: {:?}",
+                                result
+                            );
+                        }
+                    }
+                }
+
+                thread::sleep(interval);
+            }
+
+            eprintln!(
+                "[RustInputAdapter] Thread exiting after {} values",
+                counter.load(Ordering::SeqCst)
+            );
+        });
+
+        self.thread_handle = Some(handle);
     }
 
     /// Stop the adapter.
     pub fn stop(&mut self) {
         eprintln!(
-            "[RustInputAdapter] Stopped after {} values",
+            "[RustInputAdapter] Stopping after {} values",
             self.counter.load(Ordering::SeqCst)
         );
 
         self.running.store(false, Ordering::SeqCst);
+
+        // Wait for the thread to finish
+        if let Some(handle) = self.thread_handle.take() {
+            if let Err(e) = handle.join() {
+                eprintln!("[RustInputAdapter] Thread join error: {:?}", e);
+            }
+        }
     }
 }
 
diff --git a/examples/05_cpp/5_c_api_adapter_rust/src/output_adapter.rs b/examples/05_cpp/5_c_api_adapter_rust/src/output_adapter.rs
@@ -3,17 +3,21 @@
 //! This module demonstrates how to implement a CSP output adapter in Rust
 //! using the C API FFI bindings.
 //!
-//! Note: The actual value retrieval requires the CSP C API symbols to be available
-//! at runtime. When built standalone, the execute callback logs that it was invoked
-//! but cannot access the actual values.
+//! The adapter receives values from the CSP graph and prints them to stderr,
+//! demonstrating how to use the C API to retrieve typed values.
 
-use std::ffi::c_void;
+use std::ffi::{c_char, c_void};
+use std::slice;
+use std::str;
 
 use crate::bindings::{
-    CCspDateTime, CCspEngineHandle, CCspInputHandle,
+    ccsp_engine_now, ccsp_input_get_last_bool, ccsp_input_get_last_datetime,
+    ccsp_input_get_last_double, ccsp_input_get_last_int64, ccsp_input_get_last_string,
+    ccsp_input_get_type, CCspDateTime, CCspEngineHandle, CCspErrorCode, CCspInputHandle,
+    CCspType,
 };
 
-/// Output adapter that prints values to stdout.
+/// Output adapter that prints values to stderr.
 ///
 /// This adapter is called by CSP whenever the input time series ticks.
 /// It retrieves the latest value and prints it with an optional prefix.
@@ -54,24 +58,65 @@ impl RustOutputAdapter {
     /// # Safety
     ///
     /// The engine and input handles must be valid.
-    ///
-    /// Note: In a full implementation with CSP C API symbols linked, this would
-    /// call ccsp_engine_now, ccsp_input_get_type, and ccsp_input_get_last_* to
-    /// retrieve and print the actual values.
     pub unsafe fn execute(&mut self, engine: CCspEngineHandle, input: CCspInputHandle) {
         let prefix = self.prefix.as_deref().unwrap_or("");
 
         self.count += 1;
 
-        // Note: Without CSP C API symbols, we can only log that execute was called.
-        // In a full implementation, we would call:
-        //   let now = ccsp_engine_now(engine);
-        //   let input_type = ccsp_input_get_type(input);
-        //   ccsp_input_get_last_int64/double/bool/string(input, &mut value)
-        eprintln!(
-            "{}[RustOutputAdapter] execute called (count={}, engine={:?}, input={:?})",
-            prefix, self.count, engine, input
-        );
+        // Get current engine time
+        let now = ccsp_engine_now(engine);
+
+        // Get the type of the input
+        let input_type = ccsp_input_get_type(input);
+
+        // Retrieve and print the value based on type
+        match input_type {
+            CCspType::Bool => {
+                let mut value: i8 = 0;
+                if ccsp_input_get_last_bool(input, &mut value) == CCspErrorCode::Ok {
+                    eprintln!(
+                        "{}[{}] bool: {}",
+                        prefix,
+                        now,
+                        if value != 0 { "true" } else { "false" }
+                    );
+                }
+            }
+            CCspType::Int64 => {
+                let mut value: i64 = 0;
+                if ccsp_input_get_last_int64(input, &mut value) == CCspErrorCode::Ok {
+                    eprintln!("{}[{}] int64: {}", prefix, now, value);
+                }
+            }
+            CCspType::Double => {
+                let mut value: f64 = 0.0;
+                if ccsp_input_get_last_double(input, &mut value) == CCspErrorCode::Ok {
+                    eprintln!("{}[{}] double: {}", prefix, now, value);
+                }
+            }
+            CCspType::String => {
+                let mut data: *const c_char = std::ptr::null();
+                let mut len: usize = 0;
+                if ccsp_input_get_last_string(input, &mut data, &mut len) == CCspErrorCode::Ok
+                    && !data.is_null() && len > 0 {
+                        let bytes = slice::from_raw_parts(data as *const u8, len);
+                        if let Ok(s) = str::from_utf8(bytes) {
+                            eprintln!("{}[{}] string: {}", prefix, now, s);
+                        } else {
+                            eprintln!("{}[{}] string: <invalid utf8>", prefix, now);
+                        }
+                    }
+            }
+            CCspType::DateTime => {
+                let mut value: CCspDateTime = 0;
+                if ccsp_input_get_last_datetime(input, &mut value) == CCspErrorCode::Ok {
+                    eprintln!("{}[{}] datetime: {} ns", prefix, now, value);
+                }
+            }
+            _ => {
+                eprintln!("{}[{}] <type {:?}>", prefix, now, input_type);
+            }
+        }
     }
 }
 
