feat: add spring_push and spring_source_row_{from_json, close}

Author: laysakura

springql.h

@@ -50,10 +50,15 @@ typedef void *SpringConfig;
 typedef void *SpringPipeline;
 
 /**
- * Row object from an in memory queue.
+ * Row object to pop from an in memory queue.
  */
 typedef void *SpringSinkRow;
 
+/**
+ * Row object to push into an in memory queue.
+ */
+typedef void *SpringSourceRow;
+
 /**
  * Returns default configuration.
  *
@@ -165,14 +170,50 @@ SpringSinkRow *spring_pop_non_blocking(const SpringPipeline *pipeline,
                                        bool *is_err);
 
 /**
- * Frees heap occupied by a `SpringRow`.
+ * Push a row into an in memory queue. This is a non-blocking function.
+ *
+ * # Returns
+ *
+ * - `Ok`: on success.
+ * - `Unavailable`: queue named `queue` does not exist.
+ */
+enum SpringErrno spring_push(const SpringPipeline *pipeline,
+                             const char *queue,
+                             const SpringSourceRow *row);
+
+/**
+ * Create a source row from JSON string
+ *
+ * # Returns
+ *
+ * - non-NULL: Successfully created a row.
+ * - NULL: Error occurred.
+ *
+ * # Errors
+ *
+ * - `InvalidFormat`: JSON string is invalid.
+ */
+SpringSourceRow *spring_source_row_from_json(const char *json);
+
+/**
+ * Frees heap occupied by a `SpringSourceRow`.
+ *
+ * # Returns
+ *
+ * - `Ok`: on success.
+ * - `CNull`: `pipeline` is a NULL pointer.
+ */
+enum SpringErrno spring_source_row_close(SpringSourceRow *row);
+
+/**
+ * Frees heap occupied by a `SpringSinkRow`.
  *
  * # Returns
  *
  * - `Ok`: on success.
  * - `CNull`: `pipeline` is a NULL pointer.
  */
-enum SpringErrno spring_row_close(SpringSinkRow *row);
+enum SpringErrno spring_sink_row_close(SpringSinkRow *row);
 
 /**
  * Get a 2-byte integer column.

src/lib.rs

@@ -10,6 +10,7 @@ pub mod spring_errno;
 pub mod spring_last_err;
 mod spring_pipeline;
 mod spring_sink_row;
+mod spring_source_row;
 
 use std::{
     ffi::{c_void, CStr},
@@ -25,6 +26,7 @@ use crate::{
     spring_last_err::{update_last_error, LastError},
     spring_pipeline::SpringPipeline,
     spring_sink_row::SpringSinkRow,
+    spring_source_row::SpringSourceRow,
 };
 use ::springql::{error::SpringError, SpringPipeline as Pipeline};
 
@@ -212,14 +214,72 @@ pub unsafe extern "C" fn spring_pop_non_blocking(
     }
 }
 
-/// Frees heap occupied by a `SpringRow`.
+/// Push a row into an in memory queue. This is a non-blocking function.
+///
+/// # Returns
+///
+/// - `Ok`: on success.
+/// - `Unavailable`: queue named `queue` does not exist.
+#[no_mangle]
+pub unsafe extern "C" fn spring_push(
+    pipeline: *const SpringPipeline,
+    queue: *const c_char,
+    row: *const SpringSourceRow,
+) -> SpringErrno {
+    let pipeline = (*pipeline).as_pipeline();
+    let queue = CStr::from_ptr(queue).to_string_lossy().into_owned();
+    let source_row = (*row).to_row();
+    let result = with_catch(|| pipeline.push(&queue, source_row));
+    match result {
+        Ok(()) => SpringErrno::Ok,
+        Err(e) => e,
+    }
+}
+
+/// Create a source row from JSON string
+///
+/// # Returns
+///
+/// - non-NULL: Successfully created a row.
+/// - NULL: Error occurred.
+///
+/// # Errors
+///
+/// - `InvalidFormat`: JSON string is invalid.
+#[no_mangle]
+pub unsafe extern "C" fn spring_source_row_from_json(json: *const c_char) -> *mut SpringSourceRow {
+    let json = CStr::from_ptr(json).to_string_lossy().into_owned();
+    let res_source_row = with_catch(|| ::springql::SpringSourceRow::from_json(&json));
+    match res_source_row {
+        Ok(source_row) => SpringSourceRow::new(source_row).into_ptr(),
+        Err(_) => ptr::null_mut(),
+    }
+}
+
+/// Frees heap occupied by a `SpringSourceRow`.
+///
+/// # Returns
+///
+/// - `Ok`: on success.
+/// - `CNull`: `pipeline` is a NULL pointer.
+#[no_mangle]
+pub extern "C" fn spring_source_row_close(row: *mut SpringSourceRow) -> SpringErrno {
+    if row.is_null() {
+        SpringErrno::CNull
+    } else {
+        SpringSourceRow::drop(row);
+        SpringErrno::Ok
+    }
+}
+
+/// Frees heap occupied by a `SpringSinkRow`.
 ///
 /// # Returns
 ///
 /// - `Ok`: on success.
 /// - `CNull`: `pipeline` is a NULL pointer.
 #[no_mangle]
-pub extern "C" fn spring_row_close(row: *mut SpringSinkRow) -> SpringErrno {
+pub extern "C" fn spring_sink_row_close(row: *mut SpringSinkRow) -> SpringErrno {
     if row.is_null() {
         SpringErrno::CNull
     } else {

src/spring_sink_row.rs

@@ -3,7 +3,8 @@
 use ::springql::SpringSinkRow as SinkRow;
 
 use std::{ffi::c_void, mem};
-/// Row object from an in memory queue.
+
+/// Row object to pop from an in memory queue.
 #[non_exhaustive]
 #[repr(transparent)]
 pub struct SpringSinkRow(*mut c_void);

src/spring_source_row.rs

@@ -0,0 +1,31 @@
+// This file is part of https://github.com/SpringQL/SpringQL-client-c which is licensed under MIT OR Apache-2.0. See file LICENSE-MIT or LICENSE-APACHE for full license details.
+
+use ::springql::SpringSourceRow as SourceRow;
+
+use std::{ffi::c_void, mem};
+
+/// Row object to push into an in memory queue.
+#[non_exhaustive]
+#[repr(transparent)]
+pub struct SpringSourceRow(*mut c_void);
+
+impl SpringSourceRow {
+    pub fn new(source_row: SourceRow) -> Self {
+        SpringSourceRow(unsafe { mem::transmute(Box::new(source_row)) })
+    }
+
+    pub fn to_row(&self) -> SourceRow {
+        unsafe { &*(self.0 as *const SourceRow) }.clone()
+    }
+
+    pub fn drop(ptr: *mut SpringSourceRow) {
+        let outer = unsafe { Box::from_raw(ptr) };
+        let inner = unsafe { Box::from_raw(outer.0) };
+        drop(inner);
+        drop(outer);
+    }
+
+    pub fn into_ptr(self) -> *mut SpringSourceRow {
+        Box::into_raw(Box::new(self))
+    }
+}