kxkdb

Kdb+ interface for the Rust programming language.

The interface comprises two features: - IPC: Connecting Rust and kdb+ processes via IPC - API: Embedding Rust code inside kdb+ processes

Documentation

Documentation for this interface can be found at https://docs.rs/kxkdb/.

Kdbplus

The kxkdb interface is forked from the excellent kdbplus interface, developed by diamondrod.

IPC

The IPC feature enables qipc communication between Rust and kdb+.

Connectivity is via TCP or Unix Domain Sockets, with support for both compression and TLS encryption of messages.

Connection and listener methods are provided, enabling development of both - Rust IPC clients of kdb+ server processes - Rust IPC servers of kdb+ client processes

Installation

Add kxkdb as a dependency, with feature ipc. You may also want to add an asynchronous runtime such as Tokio.

e.g. toml [dependencies] kxkdb = { version = "0.0", features = ["ipc"] } tokio = { version = "1.24", features = ["full"] }

Examples

Client

```rust use kxkdb::ipc::*; use kxkdb::qattribute;

[tokio::main]

async fn main() -> Result<()> { let mut socket; // socket connection to kdb+ process let mut result; // result of sync query to kdb+ process let mut message; // compound list containing message

// connect via UDS to local kdb+ process listening on port 4321
socket = QStream::connect(ConnectionMethod::UDS, "", 4321_u16, "").await?;

// confirm connection type
println!("Connection type: {}", socket.get_connection_type());

// synchronously query kdb+ process using string
result = socket.send_sync_message(&"sum 1+til 100").await?;
println!("result1: {}", result);

// asynchronously define function in kdb+ process
socket.send_async_message(&"add_one:{x+1}").await?;

// synchronously call function (correctly)
result = socket.send_sync_message(&"add_one 41").await?;
println!("result2: {}", result);

// synchronously call function (incorrectly)
result = socket.send_sync_message(&"add_one`41").await?;
println!("result3: {}", result);

// synchronously query kdb+ process using compound list
message = K::new_compound_list(vec![K::new_symbol(String::from("add_one")), K::new_long(100)]);
result = socket.send_sync_message(&message).await?;
println!("result4: {}", result);

// asynchronously call show function in kdb+ process
message = K::new_compound_list(vec![K::new_string(String::from("show"), qattribute::NONE), K::new_symbol(String::from("hello from rust"))]);
socket.send_async_message(&message).await?;

// close socket
socket.shutdown().await?;

Ok(())

} ```

Server

Setup a credentials file containing usernames and (SHA-1 encrypted) passwords.

e.g. $ cat userpass.txt fred:e962cde7053eed120f928cd18e58ebd31be77543 homer:df43ad44d44e898f8f4e6ed91e6952bfce573e12 Note: Hashed passwords can be generated in q using .Q.sha1.

Store the path of this file in environment variable KDBPLUSACCOUNTFILE.

e.g. $ export KDBPLUS_ACCOUNT_FILE=`pwd`/userpass.txt

The following code will establish a Rust server process, listening on port 4321. ```rust use kxkdb::ipc::*;

[tokio::main]

async fn main() -> Result<()> { let mut socket; // socket connection to kdb+ process

// listen for incoming TCP connections on port 4321
socket = QStream::accept(ConnectionMethod::TCP, "127.0.0.1", 4321).await?;

// when a connection is established, synchronously send a message to the client
let response = socket.send_sync_message(&"0N!string `Hello").await?;
println!("result: {}", response);

// close socket
socket.shutdown().await?;

Ok(())

}

```

A kdb+ client can then connect using the correct credentials.

e.g. q q)hopen`:127.0.0.1:4321:fred:flintstone; "Hello"

Type Mapping

The following table displays the input types used to construct different q types (implemented as K objects).

| q | Rust | |------------------|-------------------------------------------| | boolean | bool | | guid | [u8; 16] | | byte | u8 | | short | i16 | | int | i32 | | long | i64 | | real | f32 | | float | f64 | | char | char | | symbol | String | | timestamp | chrono::DateTime<Utc> | | month | chrono::NaiveDate | | date | chrono::NaiveDate | | datetime | chrono::DateTime<Utc> | | timespan | chrono::Duration | | minute | chrono::Duration | | second | chrono::Duration | | time | chrono::Duration | | list | Vec<T> (T a corresponding type above) | | compound list | Vec<K> | | table | Vec<K> | | dictionary | Vec<K> | | generic null | () |

Note: The input type can differ from the inner type. For example, timestamp has an input type of chrono::DateTime<Utc> but the inner type is i64, denoting an elapsed time in nanoseconds since 2000.01.01D00:00:00.

Environment Variables

KDBPLUSACCOUNTFILE

Path to a credential file, used by a Rust server to manage access from kdb+ clients.

Contains a user name and SHA-1 hashed password on each line, delimited by ':'.

KDBPLUSTLSKEY_FILE

The path to a pkcs12 file used for TLS connections.

KDBPLUSTLSKEYFILESECRET

The password for the above pkcs12 file.

QUDSPATH

Defines the (real or abstract) path used for Unix Domain Sockets to $QUDSPATH/kx.[PORT].

n.b. If not defined, this will default to /tmp/kx.[PORT]

API

The API feature enables the development of shared object libraries in Rust, which can be dynamically loaded into kdb+.

In order to avoid large unsafe blocks, most native C API functions are provided with a wrapper funtion and with intuitive implementation as a trait method. The exceptions are variadic functions knk and k, which are provided under native namespace with the other C API functions.

Installation

Add kxkdb as a dependency, with feature api.

toml [dependencies] kxkdb={version="0.0", features=["api"]}

Examples

C API Style

```rust use kxkdb::qtype; use kxkdb::api::; use kxkdb::api::native::;

[no_mangle]

pub extern "C" fn createsymbollist(: K) -> K { unsafe{ let mut list=ktn(qtype::SYMBOLLIST as i32, 0); js(&mut list, ss(strtoS!("Abraham"))); js(&mut list, ss(strtoS!("Isaac"))); js(&mut list, ss(strtoS!("Jacob"))); js(&mut list, sn(strtoS!("Josephine"), 6)); list } }

[no_mangle]

pub extern "C" fn catchy(func: K, args: K) -> K { unsafe{ let result=ee(dot(func, args)); if (result).qtype == qtype::ERROR{ println!("error: {}", S_to_str((result).value.symbol)); // Decrement reference count of the error object r0(result); KNULL } else { result } } }

[no_mangle]

pub extern "C" fn dictionarylisttotable() -> K { unsafe{ let dicts = knk(3); let dictsslice = dicts.asmutslice::(); for i in 0..3 { let keys = ktn(qtype::SYMBOLLIST as i32, 2); let keysslice = keys.asmutslice::(); keysslice[0] = ss(strtoS!("a")); keysslice[1] = ss(strtoS!("b")); let values = ktn(qtype::INTLIST as i32, 2); values.asmutslice::()[0..2].copyfromslice(&[i*10, i*100]); dictsslice[i as usize] = xD(keys, values); } // Format list of dictionary as a table. // ([] a: 0 10 20i; b: 0 100 200i) k(0, strtoS!("{[dicts] -1 _ dicts, (::)}"), dicts, KNULL) } } ```

// Build values
let values = new_list(qtype::COMPOUND_LIST, 2);
let time = new_list(qtype::TIMESTAMP_LIST, 3);
// 2003.10.10D02:24:19.167018272 2006.05.24D06:16:49.419710368 2008.08.12D23:12:24.018691392
time.as_mut_slice::<J>().copy_from_slice(&[119067859167018272_i64, 201766609419710368, 271897944018691392]);
let temperature = new_list(qtype::FLOAT_LIST, 3);
temperature.as_mut_slice::<F>().copy_from_slice(&[22.1_f64, 24.7, 30.5]);
values.as_mut_slice::<K>().copy_from_slice(&[time, temperature]);

flip(new_dictionary(keys, values))