kevy_client/
lib.rs

1//! kevy-client — unified KV facade so downstream code can switch between
2//! in-process embedded and TCP-server backends with one URL string.
3//!
4//! ```no_run
5//! use kevy_client::Connection;
6//!
7//! // Same business code regardless of backend:
8//! let mut conn = Connection::open(std::env::var("MY_KEVY_URL").unwrap().as_str())?;
9//! conn.set(b"hello", b"world")?;
10//! assert_eq!(conn.get(b"hello")?, Some(b"world".to_vec()));
11//! # Ok::<(), std::io::Error>(())
12//! ```
13//!
14//! URL schemes:
15//! - `mem://`                       — in-process embedded, in-memory only
16//! - `mem://<name>`                 — shared in-process bus keyed by `<name>`
17//! - `file:///abs/path` /
18//!   `file://./rel/path`            — in-process embedded with persistence
19//! - `kevy://host[:port][/db]`      — TCP RESP, kevy-native scheme
20//! - `redis://host[:port][/db]`     — TCP RESP, standard Redis URL (alias)
21//! - `tcp://host[:port]`            — TCP RESP, raw (no SELECT round-trip)
22//!
23//! Auth (`redis://user:pass@…`) and TLS (`rediss://`) are rejected up front
24//! — kevy ships without either. v1.1.0 added the full string/hash/list/set/
25//! zset + one-shot `PUBLISH` surface. v1.2.0 added the pub/sub *consumer*
26//! side as a separate [`Subscriber`] type — a subscribed connection cannot
27//! send normal commands, so it needs its own socket and lives outside the
28//! `Connection` enum. v1.3.0 routes `mem://<name>` / `file:///path` through
29//! a process-local registry so the publisher and consumer can find each
30//! other when both opens use the same URL. The trait-vs-enum design
31//! decision is enum for now (closed two-backend universe); see ROADMAP
32//! for the trait extension path.
33
34#![forbid(unsafe_code)]
35#![warn(missing_docs)]
36
37use std::io;
38use std::time::Duration;
39
40use kevy_embedded::Store;
41use kevy_resp::Reply;
42use kevy_resp_client::RespClient;
43
44mod cluster;
45mod cluster_coll;
46mod collections;
47mod reply;
48mod scan;
49mod subscribe;
50mod subscribe_io;
51mod transaction;
52mod url;
53
54pub use cluster::ClusterClient;
55pub use subscribe::{PubsubEvent, Subscriber, SubscriberEvents, SubscriberMessages};
56pub use transaction::{Transaction, TransactionReplies};
57
58pub(crate) use reply::{array_to_bulks, store_err, string, unexpected, vec2, vec3};
59pub(crate) use url::{Target, parse_url, resolve_store};
60
61/// One open connection to a kevy backend, opaque about whether the backend
62/// is in-process or over TCP.
63pub enum Connection {
64    /// In-process [`kevy_embedded::Store`]. Boxed because `Store` is
65    /// sizeable (carries its `Config`, including v1.20 replica
66    /// upstream/backoff fields) and dwarfs the `RespClient` variant.
67    Embedded(Box<Store>),
68    /// TCP [`kevy_resp_client::RespClient`].
69    Remote(RespClient),
70}
71
72impl Connection {
73    /// Open a backend chosen by URL scheme.
74    ///
75    /// See the crate-level docs for the supported URL forms. From v1.3.0,
76    /// two `Connection::open` calls with the same `mem://<name>` or
77    /// `file:///path` URL share the same backing `Store` — and the same
78    /// pub/sub bus, so `Connection::publish` reaches a `Subscriber::open`
79    /// opened with the same URL.
80    pub fn open(url: &str) -> io::Result<Self> {
81        let parsed = parse_url(url)?;
82        match parsed {
83            Target::Remote(remote_url) => Ok(Self::Remote(RespClient::from_url(&remote_url)?)),
84            embed => Ok(Self::Embedded(Box::new(resolve_store(&embed)?))),
85        }
86    }
87
88    /// `PING`. Returns `()` on `+PONG`, propagates any IO or RESP error.
89    /// The first thing every healthcheck calls.
90    pub fn ping(&mut self) -> io::Result<()> {
91        match self {
92            Self::Embedded(_) => Ok(()),
93            Self::Remote(c) => match c.request_borrowed(&[b"PING"])? {
94                Reply::Simple(s) if s == b"PONG" => Ok(()),
95                Reply::Error(e) => Err(io::Error::other(string(e))),
96                other => Err(unexpected(other)),
97            },
98        }
99    }
100
101    /// `SET key value`. Unconditional set (no NX/XX). Returns `()` on success.
102    pub fn set(&mut self, key: &[u8], value: &[u8]) -> io::Result<()> {
103        match self {
104            Self::Embedded(s) => s.set(key, value).map(|_| ()),
105            Self::Remote(c) => match c.request_borrowed(&[b"SET", key, value])? {
106                Reply::Simple(s) if s == b"OK" => Ok(()),
107                Reply::Error(e) => Err(io::Error::other(string(e))),
108                other => Err(unexpected(other)),
109            },
110        }
111    }
112
113    /// `GET key`. `None` if absent or expired.
114    pub fn get(&mut self, key: &[u8]) -> io::Result<Option<Vec<u8>>> {
115        match self {
116            Self::Embedded(s) => s.get(key),
117            Self::Remote(c) => match c.request_borrowed(&[b"GET", key])? {
118                Reply::Bulk(v) => Ok(Some(v)),
119                Reply::Nil => Ok(None),
120                Reply::Error(e) => Err(io::Error::other(string(e))),
121                other => Err(unexpected(other)),
122            },
123        }
124    }
125
126    /// `DEL key [key ...]`. Returns the count of keys that were actually
127    /// removed (existing + dropped). Missing keys don't contribute.
128    pub fn del(&mut self, keys: &[&[u8]]) -> io::Result<usize> {
129        match self {
130            Self::Embedded(s) => s.del(keys),
131            Self::Remote(c) => {
132                let mut args = Vec::with_capacity(keys.len() + 1);
133                args.push(b"DEL".to_vec());
134                args.extend(keys.iter().map(|k| k.to_vec()));
135                match c.request(&args)? {
136                    Reply::Int(n) if n >= 0 => Ok(n as usize),
137                    Reply::Error(e) => Err(io::Error::other(string(e))),
138                    other => Err(unexpected(other)),
139                }
140            }
141        }
142    }
143
144    /// `EXISTS key [key ...]`. Count of keys present (a single key can
145    /// contribute >1 if passed multiple times, matching Redis semantics).
146    pub fn exists(&mut self, keys: &[&[u8]]) -> io::Result<usize> {
147        match self {
148            Self::Embedded(s) => s.exists(keys),
149            Self::Remote(c) => {
150                let mut args = Vec::with_capacity(keys.len() + 1);
151                args.push(b"EXISTS".to_vec());
152                args.extend(keys.iter().map(|k| k.to_vec()));
153                match c.request(&args)? {
154                    Reply::Int(n) if n >= 0 => Ok(n as usize),
155                    Reply::Error(e) => Err(io::Error::other(string(e))),
156                    other => Err(unexpected(other)),
157                }
158            }
159        }
160    }
161
162    /// `INCR key`. Returns the post-increment value. Errors on non-integer
163    /// stored value.
164    pub fn incr(&mut self, key: &[u8]) -> io::Result<i64> {
165        match self {
166            Self::Embedded(s) => s.incr(key),
167            Self::Remote(c) => match c.request_borrowed(&[b"INCR", key])? {
168                Reply::Int(n) => Ok(n),
169                Reply::Error(e) => Err(io::Error::other(string(e))),
170                other => Err(unexpected(other)),
171            },
172        }
173    }
174
175    /// `INCRBY key delta`. Negative delta is `DECRBY`. Returns post-value.
176    pub fn incr_by(&mut self, key: &[u8], delta: i64) -> io::Result<i64> {
177        match self {
178            Self::Embedded(s) => s.incr_by(key, delta),
179            Self::Remote(c) => {
180                let args = vec![
181                    b"INCRBY".to_vec(),
182                    key.to_vec(),
183                    delta.to_string().into_bytes(),
184                ];
185                match c.request(&args)? {
186                    Reply::Int(n) => Ok(n),
187                    Reply::Error(e) => Err(io::Error::other(string(e))),
188                    other => Err(unexpected(other)),
189                }
190            }
191        }
192    }
193
194    /// `PEXPIRE key ttl_ms`. Returns whether the key existed and got a TTL.
195    pub fn expire(&mut self, key: &[u8], ttl: Duration) -> io::Result<bool> {
196        match self {
197            Self::Embedded(s) => s.expire(key, ttl),
198            Self::Remote(c) => {
199                let ms = ttl.as_millis().min(i64::MAX as u128) as i64;
200                let args = vec![b"PEXPIRE".to_vec(), key.to_vec(), ms.to_string().into_bytes()];
201                match c.request(&args)? {
202                    Reply::Int(1) => Ok(true),
203                    Reply::Int(0) => Ok(false),
204                    Reply::Error(e) => Err(io::Error::other(string(e))),
205                    other => Err(unexpected(other)),
206                }
207            }
208        }
209    }
210
211    /// `PERSIST key`. Returns whether a TTL was actually removed.
212    pub fn persist(&mut self, key: &[u8]) -> io::Result<bool> {
213        match self {
214            Self::Embedded(s) => s.persist(key),
215            Self::Remote(c) => match c.request_borrowed(&[b"PERSIST", key])? {
216                Reply::Int(1) => Ok(true),
217                Reply::Int(0) => Ok(false),
218                Reply::Error(e) => Err(io::Error::other(string(e))),
219                other => Err(unexpected(other)),
220            },
221        }
222    }
223
224    /// `PTTL key`. Returns ms remaining, -2 if no key, -1 if key has no TTL.
225    pub fn ttl_ms(&mut self, key: &[u8]) -> io::Result<i64> {
226        match self {
227            Self::Embedded(s) => Ok(s.ttl_ms(key)),
228            Self::Remote(c) => match c.request_borrowed(&[b"PTTL", key])? {
229                Reply::Int(n) => Ok(n),
230                Reply::Error(e) => Err(io::Error::other(string(e))),
231                other => Err(unexpected(other)),
232            },
233        }
234    }
235
236    /// `TYPE key`. Returns the value's type as a Redis-style string (e.g.
237    /// `"string"`, `"hash"`, `"list"`, `"set"`, `"zset"`, or `"none"` if
238    /// the key doesn't exist).
239    pub fn type_of(&mut self, key: &[u8]) -> io::Result<String> {
240        match self {
241            Self::Embedded(s) => Ok(s.type_of(key).to_string()),
242            Self::Remote(c) => match c.request_borrowed(&[b"TYPE", key])? {
243                Reply::Simple(s) => Ok(string(s)),
244                Reply::Error(e) => Err(io::Error::other(string(e))),
245                other => Err(unexpected(other)),
246            },
247        }
248    }
249
250    /// `DBSIZE`. Total live keys at the time of the call.
251    pub fn dbsize(&mut self) -> io::Result<usize> {
252        match self {
253            Self::Embedded(s) => Ok(s.dbsize()),
254            Self::Remote(c) => match c.request_borrowed(&[b"DBSIZE"])? {
255                Reply::Int(n) if n >= 0 => Ok(n as usize),
256                Reply::Error(e) => Err(io::Error::other(string(e))),
257                other => Err(unexpected(other)),
258            },
259        }
260    }
261
262    /// `FLUSHALL`. Drops every key. Persistence remains opted-in; embedded
263    /// `with_persist` will rewrite the AOF on its next sync cycle.
264    ///
265    /// Named `flushall` — **not** `flush` — to avoid colliding with
266    /// `Write::flush`'s "sync buffered writes to disk" meaning; this WIPES the
267    /// store rather than persisting it.
268    pub fn flushall(&mut self) -> io::Result<()> {
269        match self {
270            Self::Embedded(s) => s.flushall(),
271            Self::Remote(c) => match c.request_borrowed(&[b"FLUSHALL"])? {
272                Reply::Simple(s) if s == b"OK" => Ok(()),
273                Reply::Error(e) => Err(io::Error::other(string(e))),
274                other => Err(unexpected(other)),
275            },
276        }
277    }
278
279    /// Deprecated alias for [`Self::flushall`]. The old name read like
280    /// `Write::flush` (sync-to-disk) but actually WIPES the store.
281    #[deprecated(
282        since = "1.8.0",
283        note = "renamed to `flushall`: `flush` collides with Write::flush (sync-to-disk); this WIPES the store"
284    )]
285    pub fn flush(&mut self) -> io::Result<()> {
286        self.flushall()
287    }
288
289    /// `SET key value PX ttl_ms`. Convenience for the common
290    /// "cache with expiry" pattern; equivalent to `set` + `expire` but
291    /// atomic.
292    pub fn set_with_ttl(&mut self, key: &[u8], value: &[u8], ttl: Duration) -> io::Result<()> {
293        match self {
294            Self::Embedded(s) => s.set_with_ttl(key, value, ttl).map(|_| ()),
295            Self::Remote(c) => {
296                let ms = ttl.as_millis().min(i64::MAX as u128) as i64;
297                let args = vec![
298                    b"SET".to_vec(),
299                    key.to_vec(),
300                    value.to_vec(),
301                    b"PX".to_vec(),
302                    ms.to_string().into_bytes(),
303                ];
304                match c.request(&args)? {
305                    Reply::Simple(s) if s == b"OK" => Ok(()),
306                    Reply::Error(e) => Err(io::Error::other(string(e))),
307                    other => Err(unexpected(other)),
308                }
309            }
310        }
311    }
312
313    /// `MGET key [key ...]` — one reply per key, `None` for missing /
314    /// wrong-type. Returns in the same order as `keys`.
315    pub fn mget(&mut self, keys: &[&[u8]]) -> io::Result<Vec<Option<Vec<u8>>>> {
316        match self {
317            Self::Embedded(s) => keys.iter().map(|k| s.get(k)).collect(),
318            Self::Remote(c) => {
319                let mut args = Vec::with_capacity(keys.len() + 1);
320                args.push(b"MGET".to_vec());
321                args.extend(keys.iter().map(|k| k.to_vec()));
322                match c.request(&args)? {
323                    Reply::Array(items) => items
324                        .into_iter()
325                        .map(|r| match r {
326                            Reply::Bulk(v) => Ok(Some(v)),
327                            Reply::Nil => Ok(None),
328                            other => Err(unexpected(other)),
329                        })
330                        .collect(),
331                    Reply::Error(e) => Err(io::Error::other(string(e))),
332                    other => Err(unexpected(other)),
333                }
334            }
335        }
336    }
337
338    /// `MSET key value [key value ...]` — set every pair atomically.
339    pub fn mset(&mut self, pairs: &[(&[u8], &[u8])]) -> io::Result<()> {
340        match self {
341            Self::Embedded(s) => {
342                for (k, v) in pairs {
343                    s.set(k, v)?;
344                }
345                Ok(())
346            }
347            Self::Remote(c) => {
348                let mut args = Vec::with_capacity(pairs.len() * 2 + 1);
349                args.push(b"MSET".to_vec());
350                for (k, v) in pairs {
351                    args.push(k.to_vec());
352                    args.push(v.to_vec());
353                }
354                match c.request(&args)? {
355                    Reply::Simple(s) if s == b"OK" => Ok(()),
356                    Reply::Error(e) => Err(io::Error::other(string(e))),
357                    other => Err(unexpected(other)),
358                }
359            }
360        }
361    }
362
363    /// `PUBLISH channel message`. Returns the count of subscribers
364    /// that received the message.
365    ///
366    /// As of v1.3.0, the embedded backend has a real in-process pub/sub
367    /// bus: when a [`Subscriber`] is open against the same `mem://<name>`
368    /// or `file:///path` URL, this delivers there and returns the actual
369    /// receiver count. Anonymous `mem://` keeps the old "no subscribers,
370    /// returns 0" behaviour (the URL is its own bus, by design).
371    ///
372    /// The pub/sub *consumer* side lives in [`Subscriber`]. On the remote
373    /// backend a subscribed TCP connection cannot send normal commands
374    /// per the RESP spec; the embedded backend has no such restriction
375    /// but `Subscriber` is still a distinct type for API symmetry.
376    pub fn publish(&mut self, channel: &[u8], message: &[u8]) -> io::Result<usize> {
377        match self {
378            Self::Embedded(s) => Ok(s.publish(channel, message)),
379            Self::Remote(c) => match c.request_borrowed(&[b"PUBLISH", channel, message])? {
380                Reply::Int(n) if n >= 0 => Ok(n as usize),
381                Reply::Error(e) => Err(io::Error::other(string(e))),
382                other => Err(unexpected(other)),
383            },
384        }
385    }
386}
387
388
389#[cfg(test)]
390mod tests {
391    use super::*;
392
393    /// Smoke against the embedded backend: every generic + string method
394    /// delegating to `Store`. Per-collection coverage (hash/list/set/zset)
395    /// lives in `collections::tests`.
396    #[test]
397    fn embedded_mem_full_crud_round_trip() {
398        let mut c = Connection::open("mem://").unwrap();
399        c.ping().unwrap();
400
401        c.set(b"k", b"v").unwrap();
402        assert_eq!(c.get(b"k").unwrap(), Some(b"v".to_vec()));
403
404        assert_eq!(c.del(&[&b"k"[..], &b"missing"[..]]).unwrap(), 1);
405        assert_eq!(c.get(b"k").unwrap(), None);
406
407        c.set(b"a", b"1").unwrap();
408        c.set(b"b", b"2").unwrap();
409        assert_eq!(c.exists(&[&b"a"[..], &b"b"[..], &b"none"[..]]).unwrap(), 2);
410
411        assert_eq!(c.incr(b"counter").unwrap(), 1);
412        assert_eq!(c.incr_by(b"counter", 9).unwrap(), 10);
413
414        c.set(b"timed", b"x").unwrap();
415        assert!(c.expire(b"timed", Duration::from_mins(1)).unwrap());
416        let ttl = c.ttl_ms(b"timed").unwrap();
417        assert!((0..=60_000).contains(&ttl), "ttl_ms = {ttl}");
418        assert!(c.persist(b"timed").unwrap());
419        assert_eq!(c.ttl_ms(b"timed").unwrap(), -1);
420
421        assert_eq!(c.type_of(b"none").unwrap(), "none");
422        assert_eq!(c.type_of(b"timed").unwrap(), "string");
423
424        assert!(c.dbsize().unwrap() >= 3);
425        c.flushall().unwrap();
426        assert_eq!(c.dbsize().unwrap(), 0);
427
428        c.set_with_ttl(b"timed2", b"x", Duration::from_mins(1))
429            .unwrap();
430        let ttl = c.ttl_ms(b"timed2").unwrap();
431        assert!((0..=60_000).contains(&ttl));
432    }
433
434    #[test]
435    fn anonymous_mem_publish_returns_zero() {
436        // No bus, no subscribers — by design.
437        let mut c = Connection::open("mem://").unwrap();
438        assert_eq!(c.publish(b"chan", b"hi").unwrap(), 0);
439    }
440
441    #[test]
442    fn embedded_mget_mset() {
443        let mut c = Connection::open("mem://").unwrap();
444        c.mset(&[
445            (b"a".as_ref(), b"1".as_ref()),
446            (b"b".as_ref(), b"2".as_ref()),
447        ])
448        .unwrap();
449        let got = c.mget(&[&b"a"[..], &b"b"[..], &b"missing"[..]]).unwrap();
450        assert_eq!(
451            got,
452            vec![Some(b"1".to_vec()), Some(b"2".to_vec()), None]
453        );
454    }
455
456    #[test]
457    fn embedded_multi_rejected_unsupported() {
458        let mut c = Connection::open("mem://").unwrap();
459        let err = c.multi().unwrap_err();
460        assert_eq!(err.kind(), io::ErrorKind::Unsupported);
461    }
462}
kevy_client/lib.rs

kevy_client/
lib.rs
