release v3.2.3 to github

Author: YutongGong

Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "dolphindb"
-version = "3.2.2"
+version = "3.2.3"
 edition = "2021"
 license = "MIT OR Apache-2.0"
 description = "A Rust native API for DolphinDB"

examples/table_insert.rs

@@ -0,0 +1,58 @@
+use dolphindb::{
+    client::ClientBuilder,
+    error::Error,
+    types::{ConstantImpl, DoubleArrayVector, Int, IntVector, TableBuilder, VectorImpl},
+};
+
+async fn table_writer() -> Result<(), Error> {
+    let mut builder = ClientBuilder::new("127.0.0.1:8848");
+    builder.with_auth(("admin", "123456"));
+    let mut client = builder.connect().await.unwrap();
+
+    let mut prices = DoubleArrayVector::new();
+    let price1 = vec![1.1, 2.2, 3.3];
+    prices.push(price1);
+    println!("{prices}");
+
+    // write one row
+    let c_int = ConstantImpl::from(Int::new(1));
+    let c_double_array_vector = ConstantImpl::Vector(VectorImpl::from(prices.clone()));
+    let res = client
+        .run_function("tableInsert{testTable}", &[c_int, c_double_array_vector])
+        .await
+        .unwrap()
+        .unwrap();
+    println!("{res}");
+
+    // write a table
+    let price2 = vec![4.4, 5.5];
+    prices.push(price2);
+    let v_int = IntVector::from_raw(&[2, 3]).into();
+    let v_double_array_vector = VectorImpl::from(prices);
+    println!("{v_double_array_vector}");
+    let mut builder = TableBuilder::new();
+    builder.with_name("my_table".to_string());
+    builder.with_contents(
+        vec![v_int, v_double_array_vector],
+        vec!["volume".to_string(), "price".to_string()],
+    );
+    let table = builder.build().unwrap();
+    println!("{table}");
+    let res = client
+        .run_function("tableInsert{testTable}", &[table.into()])
+        .await?
+        .unwrap();
+    println!("{res}");
+    Ok(())
+}
+
+#[tokio::main]
+async fn main() {
+    /*
+       DolphinDB script for creating testTable:
+       ```
+       share table(1:0, `a`b, "INT" "DOUBLE[]") as testTable
+       ```
+    */
+    table_writer().await.unwrap();
+}

examples/table_writer.rs

@@ -1,52 +1,112 @@
+use chrono::{NaiveDateTime, Utc};
 use dolphindb::{
-    client::ClientBuilder,
-    error::Error,
-    types::{ConstantImpl, DoubleArrayVector, Int, IntVector, TableBuilder, VectorImpl},
+    client::{ClientBuilder, TableWriter},
+    types::PrimitiveType,
 };
+use tokio::{
+    sync::mpsc,
+    time::{sleep, Duration},
+};
+
+#[derive(Clone)]
+struct TickerEvent {
+    event_time: i64,
+    event_time2: NaiveDateTime,
+    symbol: String,
+    event_id: i64,
+    prices: Vec<f64>,
+}
+
+fn build_table_row(event: &TickerEvent) -> Vec<PrimitiveType> {
+    vec![
+        event.event_time.into(),
+        event.event_time2.into(),
+        event.symbol.clone().into(),
+        event.event_id.into(),
+        event.prices.clone().into(),
+    ]
+}
 
-async fn table_writer() -> Result<(), Error> {
+#[tokio::main]
+async fn main() {
+    // connect to DolphinDB
     let mut builder = ClientBuilder::new("127.0.0.1:8848");
     builder.with_auth(("admin", "123456"));
     let mut client = builder.connect().await.unwrap();
 
-    let mut prices = DoubleArrayVector::new();
-    let price1 = vec![1.1, 2.2, 3.3];
-    prices.push(price1);
-    println!("{prices}");
-
-    // write one row
-    let c_int = ConstantImpl::from(Int::new(1));
-    let c_double_array_vector = ConstantImpl::Vector(VectorImpl::from(prices.clone()));
-    let res = client
-        .run_function("tableInsert{testTable}", &[c_int, c_double_array_vector])
-        .await
-        .unwrap()
-        .unwrap();
-    println!("{res}");
-
-    // write a table
-    let price2 = vec![4.4, 5.5];
-    prices.push(price2);
-    let v_int = IntVector::from_raw(&[2, 3]).into();
-    let v_double_array_vector = VectorImpl::from(prices);
-    println!("{v_double_array_vector}");
-    let mut builder = TableBuilder::new();
-    builder.with_name("my_table".to_string());
-    builder.with_contents(
-        vec![v_int, v_double_array_vector],
-        vec!["volume".to_string(), "price".to_string()],
+    // create a stream table
+    let stream_table = "depthStreamTable";
+    let script = format!(
+        r#"
+        colNames = ["event_time", "event_time2", "symbol", "event_id", "prices"]
+        colTypes = [TIMESTAMP, TIMESTAMP, SYMBOL, LONG, DOUBLE[]]
+
+        if (!existsStreamTable("{stream_table}")) {{
+            enableTableShareAndCachePurge(streamTable(1000000:0, colNames, colTypes), "{stream_table}", 1000000)
+        }}
+    "#
     );
-    let table = builder.build().unwrap();
-    println!("{table}");
-    let res = client
-        .run_function("tableInsert{testTable}", &[table.into()])
-        .await?
-        .unwrap();
-    println!("{res}");
-    Ok(())
-}
+    client.run_script(&script).await.unwrap();
 
-#[tokio::main]
-async fn main() {
-    table_writer().await.unwrap();
+    // generate data in rust
+    let event = TickerEvent {
+        event_time: Utc::now().timestamp_millis(),
+        event_time2: Utc::now().naive_utc(),
+        symbol: "BTCUSDT".into(),
+        event_id: 1000,
+        prices: vec![5000.0; 100],
+    };
+
+    // TableWriter is NOT thread safe.
+    // This example show how to insert data from multiple sources.
+    let (tx, mut rx) = mpsc::unbounded_channel::<TickerEvent>();
+    let symbol_number = 500;
+    let tx1 = tx.clone();
+    let tx2 = tx.clone();
+    let event1 = event.clone();
+    let event2 = event.clone();
+
+    // This data source generates 500 * 20 rows every second.
+    tokio::spawn(async move {
+        loop {
+            for _ in 0..symbol_number {
+                let _ = tx1.send(event1.clone());
+            }
+            sleep(Duration::from_millis(1000 / 20)).await;
+        }
+    });
+    // This data source generates 500 * 10 rows every second.
+    tokio::spawn(async move {
+        loop {
+            for _ in 0..symbol_number {
+                let _ = tx2.send(event2.clone());
+            }
+            sleep(Duration::from_millis(1000 / 10)).await;
+        }
+    });
+
+    tokio::spawn(async move {
+        let mut inserted = 0usize;
+        let mut writer = TableWriter::new(client, stream_table, 512).await;
+        while let Some(event) = rx.recv().await {
+            let mut row = build_table_row(&event);
+            let res = writer.append_row(&mut row).await;
+
+            match res {
+                Ok(_) => {
+                    inserted += 1;
+                    if inserted % 10000 == 0 {
+                        println!("{} rows inserted and {} rows in buffer", inserted, rx.len());
+                    }
+                }
+                Err(e) => {
+                    eprintln!("Insertion failed: {:?}", e);
+                    inserted += 1;
+                }
+            }
+        }
+    });
+
+    println!("Insertion started.");
+    futures::future::pending::<()>().await;
 }

src/client/mod.rs

@@ -4,6 +4,7 @@
 
 mod builder;
 mod request_info;
+mod table_writer;
 use bytes::BytesMut;
 pub(crate) use request_info::*;
 use std::collections::HashMap;
@@ -12,6 +13,7 @@ use tokio::io::{AsyncWriteExt, BufReader};
 use tokio::net::tcp::{OwnedReadHalf, OwnedWriteHalf};
 
 pub use builder::ClientBuilder;
+pub use table_writer::TableWriter;
 
 use crate::request::BehaviorOptions;
 use crate::{