pub mod abi_aof;
pub mod abi_cmd;
pub mod abi_core;
pub mod abi_kv;
pub mod abi_pubsub;
#[cfg(test)]
#[path = "abi_tests.rs"]
mod tests;
use std::collections::BTreeMap;
use std::io::Write;
use std::sync::Mutex;
use std::sync::atomic::{AtomicU32, Ordering};
use kevy_embedded::{Store, Subscription};
use kevy_store::{KevyError, StoreError};
pub const ABI_VERSION: u32 = 1;
pub(crate) const OK: i32 = 0;
pub(crate) const ERR: i32 = -1;
pub(crate) const BAD_HANDLE: i32 = -2;
pub(crate) struct Instance {
pub(crate) store: Store,
pub(crate) subs: BTreeMap<u32, Subscription>,
pub(crate) next_sub: u32,
pub(crate) capture_aof: bool,
pub(crate) aof_out: Vec<u8>,
pub(crate) aof_in_carry: Vec<u8>,
pub(crate) aof_in_started: bool,
pub(crate) aof_format: kevy_persist::AofFormat,
pub(crate) aof_out_started: bool,
pub(crate) aof_scratch: Vec<u8>,
pub(crate) out: Vec<u8>,
}
impl Instance {
pub(crate) fn new(store: Store, capture_aof: bool) -> Self {
Instance {
store,
subs: BTreeMap::new(),
next_sub: 1,
capture_aof,
aof_out: Vec::new(),
aof_in_carry: Vec::new(),
aof_in_started: false,
aof_format: kevy_persist::AofFormat::V2,
aof_out_started: false,
aof_scratch: Vec::new(),
out: Vec::new(),
}
}
pub(crate) fn put_out(&mut self, bytes: &[u8]) {
self.out.clear();
self.out.extend_from_slice(bytes);
}
pub(crate) fn fail(&mut self, msg: impl std::fmt::Display) -> i32 {
self.out.clear();
let _ = write!(self.out, "{msg}");
ERR
}
pub(crate) fn fail_kevy(&mut self, e: &KevyError) -> i32 {
match e {
KevyError::Store(se) => self.fail(store_err_canonical(se)),
other => self.fail(other),
}
}
pub(crate) fn log_frame(&mut self, parts: &[&[u8]]) {
if !self.capture_aof {
return;
}
let argv = kevy_persist::Argv::from(parts.iter().map(|p| p.to_vec()).collect::<Vec<_>>());
match self.aof_format {
kevy_persist::AofFormat::V1 => {
let _ = kevy_persist::write_multibulk(&mut self.aof_out, &argv);
}
kevy_persist::AofFormat::V2 => {
if !self.aof_out_started {
self.aof_out.extend_from_slice(kevy_persist::AOF2_MAGIC);
self.aof_out_started = true;
}
let _ = kevy_persist::write_record_multibulk(
&mut self.aof_out,
&argv,
&mut self.aof_scratch,
);
}
}
}
}
fn store_err_canonical(e: &StoreError) -> &'static str {
match e {
StoreError::WrongType => {
"WRONGTYPE Operation against a key holding the wrong kind of value"
}
StoreError::NotInteger => "ERR value is not an integer or out of range",
StoreError::Overflow => "ERR increment or decrement would overflow",
StoreError::OutOfRange => "ERR index out of range",
StoreError::NoSuchKey => "ERR no such key",
StoreError::NotFloat => "ERR value is not a valid float",
StoreError::OutOfMemory => "OOM command not allowed when used memory > 'maxmemory'.",
}
}
pub(crate) static NEXT_ID: AtomicU32 = AtomicU32::new(1);
pub(crate) static REG: Mutex<BTreeMap<u32, Instance>> = Mutex::new(BTreeMap::new());
pub(crate) fn next_id() -> u32 {
NEXT_ID.fetch_add(1, Ordering::Relaxed)
}
pub(crate) fn with<R>(h: u32, missing: R, f: impl FnOnce(&mut Instance) -> R) -> R {
let mut reg = REG.lock().unwrap_or_else(std::sync::PoisonError::into_inner);
match reg.get_mut(&h) {
Some(inst) => f(inst),
None => missing,
}
}
pub(crate) unsafe fn arg<'a>(ptr: *const u8, len: u32) -> &'a [u8] {
if len == 0 {
return &[];
}
unsafe { std::slice::from_raw_parts(ptr, len as usize) }
}