diff --git a/content/develop/use-cases/cache-aside/rust/Cargo.toml b/content/develop/use-cases/cache-aside/rust/Cargo.toml index 59b9c42c2f..0d6a9a247f 100644 --- a/content/develop/use-cases/cache-aside/rust/Cargo.toml +++ b/content/develop/use-cases/cache-aside/rust/Cargo.toml @@ -4,7 +4,7 @@ version = "0.1.0" edition = "2021" [dependencies] -redis = { version = "0.24", features = ["tokio-comp", "aio", "connection-manager"] } +redis = { version = "1.8", features = ["tokio-comp", "aio", "connection-manager"] } tokio = { version = "1", features = ["full"] } axum = "0.7" serde = { version = "1.0", features = ["derive"] } diff --git a/content/develop/use-cases/cache-aside/rust/_index.md b/content/develop/use-cases/cache-aside/rust/_index.md index f6208c5b6f..ce7dbe4faf 100644 --- a/content/develop/use-cases/cache-aside/rust/_index.md +++ b/content/develop/use-cases/cache-aside/rust/_index.md @@ -191,7 +191,7 @@ where .release_script .key(&lock_key) .arg(token.as_str()) - .invoke_async::<_, i64>(&mut conn) + .invoke_async::(&mut conn) .await; return result; } @@ -234,10 +234,10 @@ pub async fn update_field(&self, id: &str, field: &str, value: &str) -> RedisRes let cache_key = self.cache_key(id); let mut conn = self.conn.clone(); loop { - redis::cmd("WATCH").arg(&cache_key).query_async::<_, ()>(&mut conn).await?; + redis::cmd("WATCH").arg(&cache_key).query_async::<()>(&mut conn).await?; let exists: i64 = conn.exists(&cache_key).await?; if exists == 0 { - redis::cmd("UNWATCH").query_async::<_, ()>(&mut conn).await?; + redis::cmd("UNWATCH").query_async::<()>(&mut conn).await?; return Ok(false); } let result: Option<((),)> = redis::pipe() @@ -293,7 +293,7 @@ Before running the demo, make sure that: ```toml [dependencies] -redis = { version = "0.24", features = ["tokio-comp", "aio", "connection-manager"] } +redis = { version = "1.8", features = ["tokio-comp", "aio", "connection-manager"] } tokio = { version = "1", features = ["full"] } axum = "0.7" serde = { version = "1.0", features = ["derive"] } diff --git a/content/develop/use-cases/cache-aside/rust/cache.rs b/content/develop/use-cases/cache-aside/rust/cache.rs index 0c1acac282..f78b0c7791 100644 --- a/content/develop/use-cases/cache-aside/rust/cache.rs +++ b/content/develop/use-cases/cache-aside/rust/cache.rs @@ -149,7 +149,7 @@ impl RedisCache { .release_script .key(&lock_key) .arg(token.as_str()) - .invoke_async::<_, i64>(&mut conn) + .invoke_async::(&mut conn) .await; return result; } @@ -190,7 +190,7 @@ impl RedisCache { .ignore() .expire(cache_key, self.cfg.ttl as i64) .ignore() - .query_async::<_, ()>(&mut conn) + .query_async::<()>(&mut conn) .await?; } Ok(record) @@ -212,11 +212,11 @@ impl RedisCache { loop { redis::cmd("WATCH") .arg(&cache_key) - .query_async::<_, ()>(&mut conn) + .query_async::<()>(&mut conn) .await?; let exists: i64 = conn.exists(&cache_key).await?; if exists == 0 { - redis::cmd("UNWATCH").query_async::<_, ()>(&mut conn).await?; + redis::cmd("UNWATCH").query_async::<()>(&mut conn).await?; return Ok(false); } let result: Option<((),)> = redis::pipe() diff --git a/content/develop/use-cases/feature-store/rust/Cargo.toml b/content/develop/use-cases/feature-store/rust/Cargo.toml index b017cfaf14..d21178153e 100644 --- a/content/develop/use-cases/feature-store/rust/Cargo.toml +++ b/content/develop/use-cases/feature-store/rust/Cargo.toml @@ -20,7 +20,7 @@ path = "build_features_bin.rs" # demo uses. HEXPIRE/HTTL are issued via `redis::cmd("HEXPIRE")` # because the typed bindings don't ship on this client line yet — # the wire protocol is the same. -redis = { version = "0.27", features = ["tokio-comp", "aio", "connection-manager"] } +redis = { version = "1.8", features = ["tokio-comp", "aio", "connection-manager"] } tokio = { version = "1", features = ["full"] } axum = "0.7" # axum-extra ships an alternative Form extractor that uses diff --git a/content/develop/use-cases/feature-store/rust/_index.md b/content/develop/use-cases/feature-store/rust/_index.md index d36d6a270c..1ba9051d07 100644 --- a/content/develop/use-cases/feature-store/rust/_index.md +++ b/content/develop/use-cases/feature-store/rust/_index.md @@ -69,10 +69,10 @@ Two crate facts shape the helper: and they all share the same underlying connection. There's no pool to manage. * **The `redis::cmd("HEXPIRE")` builder is how you reach commands not yet - typed on the `AsyncCommands` trait.** Per-field TTL bindings (`hexpire`, - `httl`, `hpersist`) aren't part of the typed surface on redis-rs 0.27, so - the helper issues them with the generic command builder. The wire bytes - are identical to the typed form. + typed on the `AsyncCommands` trait.** The helper issues the per-field TTL + commands (`HEXPIRE`, `HTTL`, `HPERSIST`) with the generic command builder. + redis-rs 1.x also ships typed `hexpire`, `httl`, and `hpersist` methods; + the wire bytes are identical. In this example, the batch features describe a user's longer-term shape (`country_iso`, `risk_segment`, `account_age_days`, `tx_count_7d`, @@ -305,7 +305,7 @@ pub async fn update_streaming( for code in &codes { if *code != 1 { return Err(redis::RedisError::from(( - redis::ErrorKind::ResponseError, + redis::ErrorKind::Server(redis::ServerErrorKind::ResponseError), "HEXPIRE invariant violated", format!("HEXPIRE did not set every field TTL for {key}: {codes:?}"), ))); @@ -340,9 +340,9 @@ debugging and as a freshness signal in the model itself. > **HEXPIRE requires Redis 7.4 or later.** `HEXPIRE` and the field-level > TTL commands (`HTTL`, `HPERSIST`, `HEXPIREAT`, `HPEXPIRE`, `HPEXPIREAT`, > `HPTTL`, `HEXPIRETIME`, `HPEXPIRETIME`) were added in Redis 7.4. The -> demo's `Cargo.toml` pins `redis = "0.27"` and uses -> `redis::cmd("HEXPIRE")` because the typed binding doesn't ship on that -> client line yet — the wire bytes are identical. +> demo's `Cargo.toml` pins `redis = "1.8"` and issues them through +> `redis::cmd("HEXPIRE")`. redis-rs 1.x also has typed bindings for some +> of them (`hexpire`, `httl`, `hpersist`) that send the same wire bytes. ### Inference reads with HMGET @@ -357,7 +357,7 @@ pub async fn get_features( if field_names.is_empty() { return Ok(BTreeMap::new()); } let key = self.key_for(entity_id); let mut conn = self.conn.clone(); - let values: Vec> = conn.hget(&key, field_names).await?; + let values: Vec> = conn.hmget(&key, field_names).await?; let mut out = BTreeMap::new(); for (n, v) in field_names.iter().zip(values.into_iter()) { if let Some(s) = v { out.insert((*n).to_string(), s); } @@ -386,7 +386,7 @@ pub async fn batch_get_features( } let mut pipe = redis::pipe(); for id in entity_ids { - pipe.hget(self.key_for(id), field_names); + pipe.hmget(self.key_for(id), field_names); } let mut conn = self.conn.clone(); let rows: Vec>> = pipe.query_async(&mut conn).await?; @@ -400,7 +400,7 @@ under 1 ms against a local Redis. A Redis Cluster is different: a single `redis::pipe()` is bound to one connection, and a `ConnectionManager` holds one connection to one node. For batch reads on a cluster, use redis-rs's -[`cluster_async`](https://docs.rs/redis/0.27/redis/cluster_async/index.html) +[`cluster_async`](https://docs.rs/redis/latest/redis/cluster_async/index.html) client and either fan out parallel `hget` calls (the cluster client routes each one to the right shard) or, for tighter control, group entity IDs by hash slot and run one pipeline per shard in parallel. A hash tag like @@ -520,10 +520,9 @@ multiplexed socket. Endpoints: * **Redis 7.4 or later.** [`HEXPIRE`](/content/commands/hexpire.md) and [`HTTL`](/content/commands/httl.md) were added in Redis 7.4; the demo relies on per-field TTL for the mixed-staleness story. -* **Rust 1.74 or later.** The demo uses `async fn` in traits, `let-else`, - and other recent ergonomics. Earlier stable Rust may compile after small - tweaks. -* **redis-rs 0.27 or later.** The demo's `Cargo.toml` pins 0.27 with the +* **Rust 1.88 or later.** The `redis` 1.8 crate requires it, and the demo + uses `async fn` in traits, `let-else`, and other recent ergonomics. +* **redis-rs 1.8 or later.** The demo's `Cargo.toml` pins 1.8 with the `tokio-comp`, `aio`, and `connection-manager` features. Per-field TTL commands are issued via `redis::cmd("HEXPIRE")`. @@ -620,7 +619,7 @@ prevents training-serving skew. On a single Redis instance, a pipelined `HMGET` across `N` users is one round trip. A Redis Cluster is different: a single `redis::pipe()` ships through one connection to one node, so on a cluster you need redis-rs's -[`cluster_async`](https://docs.rs/redis/0.27/redis/cluster_async/index.html) +[`cluster_async`](https://docs.rs/redis/latest/redis/cluster_async/index.html) client. Either fan out parallel `hget` calls (the cluster client routes each one to the right shard) or group entity IDs by hash slot and issue one pipeline against each shard in parallel. diff --git a/content/develop/use-cases/feature-store/rust/feature_store.rs b/content/develop/use-cases/feature-store/rust/feature_store.rs index d138725880..50970bfb57 100644 --- a/content/develop/use-cases/feature-store/rust/feature_store.rs +++ b/content/develop/use-cases/feature-store/rust/feature_store.rs @@ -247,7 +247,7 @@ impl FeatureStore { for code in &codes { if *code != 1 { return Err(redis::RedisError::from(( - redis::ErrorKind::ResponseError, + redis::ErrorKind::Server(redis::ServerErrorKind::ResponseError), "HEXPIRE invariant violated", format!( "HEXPIRE did not set every field TTL for {key}: {codes:?}" @@ -278,7 +278,7 @@ impl FeatureStore { } let key = self.key_for(entity_id); let mut conn = self.conn.clone(); - let values: Vec> = conn.hget(&key, field_names).await?; + let values: Vec> = conn.hmget(&key, field_names).await?; let mut out = BTreeMap::new(); for (n, v) in field_names.iter().zip(values.into_iter()) { if let Some(s) = v { @@ -320,7 +320,7 @@ impl FeatureStore { } let mut pipe = redis::pipe(); for id in entity_ids { - pipe.hget(self.key_for(id), field_names); + pipe.hmget(self.key_for(id), field_names); } let mut conn = self.conn.clone(); let rows: Vec>> = pipe.query_async(&mut conn).await?; @@ -377,7 +377,7 @@ impl FeatureStore { // shape coercion here the way redis-py and Lettuce do for // their respective clients. let values: Value = cmd.query_async(&mut conn).await?; - let codes: Vec = Vec::::from_redis_value(&values).unwrap_or_else(|_| { + let codes: Vec = Vec::::from_redis_value_ref(&values).unwrap_or_else(|_| { // Defensive fallback: if the client ever returns nil for // a missing key, treat every field as -2. field_names.iter().map(|_| -2).collect() @@ -404,6 +404,7 @@ impl FeatureStore { let mut ids: Vec = Vec::new(); let prefix_len = self.key_prefix.len(); while let Some(k) = iter.next_item().await { + let k = k?; if k.len() > prefix_len { ids.push(k[prefix_len..].to_string()); } @@ -422,7 +423,8 @@ impl FeatureStore { let mut conn = self.conn.clone(); let mut iter: AsyncIter = conn.scan_match(&pattern).await?; let mut n = 0i64; - while let Some(_k) = iter.next_item().await { + while let Some(k) = iter.next_item().await { + k?; n += 1; } Ok(n) @@ -444,7 +446,7 @@ impl FeatureStore { let mut batch: Vec = Vec::with_capacity(500); let mut iter: AsyncIter = conn.scan_match(&pattern).await?; while let Some(k) = iter.next_item().await { - batch.push(k); + batch.push(k?); if batch.len() >= 500 { let n: i64 = self.conn.clone().del(batch.as_slice()).await?; deleted += n; diff --git a/content/develop/use-cases/job-queue/rust/Cargo.lock b/content/develop/use-cases/job-queue/rust/Cargo.lock index 3f2c6b2cf9..ffd9f97a0b 100644 --- a/content/develop/use-cases/job-queue/rust/Cargo.lock +++ b/content/develop/use-cases/job-queue/rust/Cargo.lock @@ -11,6 +11,23 @@ dependencies = [ "rustversion", ] +[[package]] +name = "arcstr" +version = "1.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "03918c3dbd7701a85c6b9887732e2921175f26c350b4563841d0958c21d57e6d" + +[[package]] +name = "async-lock" +version = "3.4.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "290f7f2596bd5b78a9fec8088ccd89180d7f9f55b94b0576823bbbdc72ee8311" +dependencies = [ + "event-listener", + "event-listener-strategy", + "pin-project-lite", +] + [[package]] name = "async-trait" version = "0.1.89" @@ -118,9 +135,9 @@ checksum = "9330f8b2ff13f34540b44e946ef35111825727b38d33286ef986142615121801" [[package]] name = "combine" -version = "4.6.7" +version = "4.6.9" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "ba5a308b75df32fe02788e748662718f03fde005016435c444eea572398219fd" +checksum = "4356eba6976435498d05a311ee9fc486da3bc81c200d3f0d7dd9703b41913f92" dependencies = [ "bytes", "futures-core", @@ -141,12 +158,6 @@ dependencies = [ "syn", ] -[[package]] -name = "either" -version = "1.15.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "48c757948c5ede0e46177b7add2e67155f70e33c07fea8284df6576da70b3719" - [[package]] name = "errno" version = "0.3.14" @@ -154,7 +165,27 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "39cab71617ae0d63f51a36d69f866391735b51691dbda63cf6f96d042b63efeb" dependencies = [ "libc", - "windows-sys 0.61.2", + "windows-sys", +] + +[[package]] +name = "event-listener" +version = "5.4.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5a23add41df1562121a9393cb065eab5146a1242410f23a644851e90cfd669d2" +dependencies = [ + "parking", + "pin-project-lite", +] + +[[package]] +name = "event-listener-strategy" +version = "0.5.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8be9f3dfaaffdae2972880079a491a1a8bb7cbed0b8dd7a347f668b4150a3b93" +dependencies = [ + "event-listener", + "pin-project-lite", ] [[package]] @@ -172,21 +203,6 @@ dependencies = [ "percent-encoding", ] -[[package]] -name = "futures" -version = "0.3.32" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "8b147ee9d1f6d097cef9ce628cd2ee62288d963e16fb287bd9286455b241382d" -dependencies = [ - "futures-channel", - "futures-core", - "futures-executor", - "futures-io", - "futures-sink", - "futures-task", - "futures-util", -] - [[package]] name = "futures-channel" version = "0.3.32" @@ -194,7 +210,6 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "07bbe89c50d7a535e539b8c17bc0b49bdb77747034daa8087407d655f3f7cc1d" dependencies = [ "futures-core", - "futures-sink", ] [[package]] @@ -203,34 +218,6 @@ version = "0.3.32" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "7e3450815272ef58cec6d564423f6e755e25379b217b0bc688e295ba24df6b1d" -[[package]] -name = "futures-executor" -version = "0.3.32" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "baf29c38818342a3b26b5b923639e7b1f4a61fc5e76102d4b1981c6dc7a7579d" -dependencies = [ - "futures-core", - "futures-task", - "futures-util", -] - -[[package]] -name = "futures-io" -version = "0.3.32" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "cecba35d7ad927e23624b22ad55235f2239cfa44fd10428eecbeba6d6a717718" - -[[package]] -name = "futures-macro" -version = "0.3.32" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "e835b70203e41293343137df5c0664546da5745f82ec9b84d40be8336958447b" -dependencies = [ - "proc-macro2", - "quote", - "syn", -] - [[package]] name = "futures-sink" version = "0.3.32" @@ -249,13 +236,9 @@ version = "0.3.32" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "389ca41296e6190b48053de0321d02a77f32f8a5d2461dd38762c0593805c6d6" dependencies = [ - "futures-channel", "futures-core", - "futures-io", - "futures-macro", "futures-sink", "futures-task", - "memchr", "pin-project-lite", "slab", ] @@ -454,15 +437,6 @@ dependencies = [ "icu_properties", ] -[[package]] -name = "itertools" -version = "0.13.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "413ee7dfc52ee1a4949ceeb7dbc8a33f2d6c088194d9f922fb8318faf1f01186" -dependencies = [ - "either", -] - [[package]] name = "itoa" version = "1.0.18" @@ -534,14 +508,14 @@ checksum = "50b7e5b27aa02a74bac8c3f23f448f8d87ff11f92d3aac1a6ed369ee08cc56c1" dependencies = [ "libc", "wasi", - "windows-sys 0.61.2", + "windows-sys", ] [[package]] name = "num-bigint" -version = "0.4.6" +version = "0.5.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "a5e44f723f1133c9deac646763579fdb3ac745e418f2a7af9cd0c431da1f20b9" +checksum = "93e7820bc0a80a0238e650327316f929ba18d5be054b647490a3a6a339f3e7c0" dependencies = [ "num-integer", "num-traits", @@ -571,6 +545,12 @@ version = "1.21.4" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "9f7c3e4beb33f85d45ae3e3a1792185706c8e16d043238c593331cc7cd313b50" +[[package]] +name = "parking" +version = "2.2.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f38d5652c16fde515bb1ecef450ab0f6a219d619a7274976324d5e377f7dceba" + [[package]] name = "parking_lot" version = "0.12.5" @@ -674,28 +654,30 @@ dependencies = [ [[package]] name = "redis" -version = "0.27.6" +version = "1.8.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "09d8f99a4090c89cc489a94833c901ead69bfbf3877b4867d5482e321ee875bc" +checksum = "7acd809e80afa754c63f586955699d8990efe1c4b6613183ee5f027a11fbc8b5" dependencies = [ "arc-swap", - "async-trait", + "arcstr", + "async-lock", "backon", "bytes", + "cfg-if", "combine", - "futures", + "futures-channel", "futures-util", - "itertools", "itoa", "num-bigint", "percent-encoding", "pin-project-lite", "ryu", "sha1_smol", - "socket2 0.5.10", + "socket2", "tokio", "tokio-util", "url", + "xxhash-rust", ] [[package]] @@ -819,16 +801,6 @@ version = "1.15.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "67b1b7a3b5fe4f1376887184045fcf45c69e92af734b7aaddc05fb777b6fbd03" -[[package]] -name = "socket2" -version = "0.5.10" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "e22376abed350d73dd1cd119b57ffccad95b4e585a7cda43e286245ce23c0678" -dependencies = [ - "libc", - "windows-sys 0.52.0", -] - [[package]] name = "socket2" version = "0.6.3" @@ -836,7 +808,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "3a766e1110788c36f4fa1c2b71b387a7815aa65f88ce0229841826633d93723e" dependencies = [ "libc", - "windows-sys 0.61.2", + "windows-sys", ] [[package]] @@ -895,9 +867,9 @@ dependencies = [ "parking_lot", "pin-project-lite", "signal-hook-registry", - "socket2 0.6.3", + "socket2", "tokio-macros", - "windows-sys 0.61.2", + "windows-sys", ] [[package]] @@ -1008,15 +980,6 @@ version = "0.2.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "f0805222e57f7521d6a62e36fa9163bc891acd422f971defe97d64e70d0a4fe5" -[[package]] -name = "windows-sys" -version = "0.52.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "282be5f36a8ce781fad8c8ae18fa3f9beff57ec1b52cb3de0789201425d9a33d" -dependencies = [ - "windows-targets", -] - [[package]] name = "windows-sys" version = "0.61.2" @@ -1026,76 +989,18 @@ dependencies = [ "windows-link", ] -[[package]] -name = "windows-targets" -version = "0.52.6" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "9b724f72796e036ab90c1021d4780d4d3d648aca59e491e6b98e725b84e99973" -dependencies = [ - "windows_aarch64_gnullvm", - "windows_aarch64_msvc", - "windows_i686_gnu", - "windows_i686_gnullvm", - "windows_i686_msvc", - "windows_x86_64_gnu", - "windows_x86_64_gnullvm", - "windows_x86_64_msvc", -] - -[[package]] -name = "windows_aarch64_gnullvm" -version = "0.52.6" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "32a4622180e7a0ec044bb555404c800bc9fd9ec262ec147edd5989ccd0c02cd3" - -[[package]] -name = "windows_aarch64_msvc" -version = "0.52.6" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "09ec2a7bb152e2252b53fa7803150007879548bc709c039df7627cabbd05d469" - -[[package]] -name = "windows_i686_gnu" -version = "0.52.6" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "8e9b5ad5ab802e97eb8e295ac6720e509ee4c243f69d781394014ebfe8bbfa0b" - -[[package]] -name = "windows_i686_gnullvm" -version = "0.52.6" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "0eee52d38c090b3caa76c563b86c3a4bd71ef1a819287c19d586d7334ae8ed66" - -[[package]] -name = "windows_i686_msvc" -version = "0.52.6" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "240948bc05c5e7c6dabba28bf89d89ffce3e303022809e73deaefe4f6ec56c66" - -[[package]] -name = "windows_x86_64_gnu" -version = "0.52.6" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "147a5c80aabfbf0c7d901cb5895d1de30ef2907eb21fbbab29ca94c5b08b1a78" - -[[package]] -name = "windows_x86_64_gnullvm" -version = "0.52.6" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "24d5b23dc417412679681396f2b49f3de8c1473deb516bd34410872eff51ed0d" - -[[package]] -name = "windows_x86_64_msvc" -version = "0.52.6" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "589f6da84c646204747d1270a2a5661ea66ed1cced2631d546fdfb155959f9ec" - [[package]] name = "writeable" version = "0.6.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "1ffae5123b2d3fc086436f8834ae3ab053a283cfac8fe0a0b8eaae044768a4c4" +[[package]] +name = "xxhash-rust" +version = "0.8.19" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "550a2b930b62486a393c52d5c3b84bff264b28aa437ed64694d31e93b1757af7" + [[package]] name = "yoke" version = "0.8.2" diff --git a/content/develop/use-cases/job-queue/rust/Cargo.toml b/content/develop/use-cases/job-queue/rust/Cargo.toml index 2d08e6cce9..2631944005 100644 --- a/content/develop/use-cases/job-queue/rust/Cargo.toml +++ b/content/develop/use-cases/job-queue/rust/Cargo.toml @@ -4,7 +4,7 @@ version = "0.1.0" edition = "2021" [dependencies] -redis = { version = "0.27", features = ["tokio-comp", "aio", "connection-manager"] } +redis = { version = "1.8", features = ["tokio-comp", "aio", "connection-manager"] } tokio = { version = "1", features = ["full"] } axum = "0.7" serde = { version = "1.0", features = ["derive"] } diff --git a/content/develop/use-cases/job-queue/rust/_index.md b/content/develop/use-cases/job-queue/rust/_index.md index 82b234b506..b6a9d33c5e 100644 --- a/content/develop/use-cases/job-queue/rust/_index.md +++ b/content/develop/use-cases/job-queue/rust/_index.md @@ -52,7 +52,7 @@ The `RedisJobQueue` struct wraps the queue operations ([source](https://github.com/redis/docs/blob/main/content/develop/use-cases/job-queue/rust/src/job_queue.rs)): ```rust -use redis::aio::ConnectionManager; +use redis::aio::{ConnectionManager, ConnectionManagerConfig}; use redis::Client; use serde_json::json; @@ -61,7 +61,10 @@ use jobqueue_demo::job_queue::{JobQueueOptions, RedisJobQueue}; #[tokio::main] async fn main() -> redis::RedisResult<()> { let client = Client::open("redis://127.0.0.1:6379/")?; - let conn = ConnectionManager::new(client).await?; + // Workers block in BLMOVE for longer than the 500 ms default response timeout + // in redis-rs 1.x, so turn the response timeout off for this connection. + let config = ConnectionManagerConfig::new().set_response_timeout(None); + let conn = ConnectionManager::new_with_config(client, config).await?; let queue = RedisJobQueue::new( conn, JobQueueOptions { @@ -161,7 +164,7 @@ pub async fn enqueue(&self, payload: Value) -> RedisResult { .ignore() .lpush(&self.pending_key, &job_id) .ignore() - .query_async::<_, ()>(&mut conn) + .query_async::<()>(&mut conn) .await?; self.enqueued_total.fetch_add(1, Ordering::Relaxed); @@ -353,14 +356,14 @@ The completed and failed lists are capped via `LTRIM` so they never grow unbound ## Prerequisites * Redis 6.2 or later running locally on the default port (6379). `BLMOVE` requires Redis 6.2+; on older servers, replace the call with `BRPOPLPUSH`. -* Rust 1.75 or later. -* The [`redis`](https://crates.io/crates/redis) crate at 0.27+ (or 0.24+ if you only need `BRPOPLPUSH`-style claims). +* Rust 1.88 or later (the `redis` 1.8 crate requires it). +* The [`redis`](https://crates.io/crates/redis) crate at 1.8 or later. Add the crate dependencies to your `Cargo.toml`: ```toml [dependencies] -redis = { version = "0.27", features = ["tokio-comp", "aio", "connection-manager"] } +redis = { version = "1.8", features = ["tokio-comp", "aio", "connection-manager"] } tokio = { version = "1", features = ["full"] } axum = "0.7" serde = { version = "1.0", features = ["derive"] } diff --git a/content/develop/use-cases/job-queue/rust/src/main.rs b/content/develop/use-cases/job-queue/rust/src/main.rs index ed650322ef..14f5e377a6 100644 --- a/content/develop/use-cases/job-queue/rust/src/main.rs +++ b/content/develop/use-cases/job-queue/rust/src/main.rs @@ -17,7 +17,7 @@ use axum::{ routing::{get, post}, Form, Json, Router, }; -use redis::aio::ConnectionManager; +use redis::aio::{ConnectionManager, ConnectionManagerConfig}; use redis::Client; use serde::Deserialize; use serde_json::{json, Value}; @@ -51,7 +51,10 @@ async fn main() { } }; - let conn = match ConnectionManager::new(client).await { + // Workers block in BLMOVE for longer than the 500 ms default response timeout + // in redis-rs 1.x, so turn the response timeout off for this connection. + let config = ConnectionManagerConfig::new().set_response_timeout(None); + let conn = match ConnectionManager::new_with_config(client, config).await { Ok(c) => c, Err(e) => { eprintln!("Failed to connect to Redis at {}: {}", redis_url, e); diff --git a/content/develop/use-cases/leaderboard/rust/Cargo.toml b/content/develop/use-cases/leaderboard/rust/Cargo.toml index 9b236afb20..8bc9d8b365 100644 --- a/content/develop/use-cases/leaderboard/rust/Cargo.toml +++ b/content/develop/use-cases/leaderboard/rust/Cargo.toml @@ -4,7 +4,7 @@ version = "0.1.0" edition = "2021" [dependencies] -redis = { version = "0.24", features = ["script", "tokio-comp"] } +redis = { version = "1.8", features = ["script", "tokio-comp"] } tokio = { version = "1", features = ["full"] } axum = "0.7" serde = { version = "1.0", features = ["derive"] } diff --git a/content/develop/use-cases/leaderboard/rust/_index.md b/content/develop/use-cases/leaderboard/rust/_index.md index d3367124e7..0c036b383c 100644 --- a/content/develop/use-cases/leaderboard/rust/_index.md +++ b/content/develop/use-cases/leaderboard/rust/_index.md @@ -48,7 +48,7 @@ Add the `redis` crate to your `Cargo.toml`: ```toml [dependencies] -redis = { version = "0.24", features = ["script", "tokio-comp"] } +redis = { version = "1.8", features = ["script", "tokio-comp"] } tokio = { version = "1", features = ["full"] } ``` diff --git a/content/develop/use-cases/leaderboard/rust/leaderboard.rs b/content/develop/use-cases/leaderboard/rust/leaderboard.rs index 7d99a1b033..414452e8b0 100644 --- a/content/develop/use-cases/leaderboard/rust/leaderboard.rs +++ b/content/develop/use-cases/leaderboard/rust/leaderboard.rs @@ -317,7 +317,7 @@ impl AsyncRedisLeaderboard { .arg(self.metadata_key(user_id)) .arg(flatten_metadata(&payload)); } - pipe.query_async::<_, ()>(con).await?; + pipe.query_async::<()>(con).await?; let trimmed_user_ids = self.trim_to_max_entries(con).await?; Ok(self @@ -553,8 +553,8 @@ fn flatten_metadata(metadata: &Metadata) -> Vec { fn parse_score_value(value: Option<&redis::Value>) -> Option { match value { Some(redis::Value::Int(v)) => Some(*v as f64), - Some(redis::Value::Data(v)) => String::from_utf8_lossy(v).parse::().ok(), - Some(redis::Value::Bulk(items)) => items.first().and_then(|value| parse_score_value(Some(value))), + Some(redis::Value::BulkString(v)) => String::from_utf8_lossy(v).parse::().ok(), + Some(redis::Value::Array(items)) => items.first().and_then(|value| parse_score_value(Some(value))), _ => None, } } diff --git a/content/develop/use-cases/memory-layer/rust/Cargo.toml b/content/develop/use-cases/memory-layer/rust/Cargo.toml index f9442800b6..5cd86c2c6a 100644 --- a/content/develop/use-cases/memory-layer/rust/Cargo.toml +++ b/content/develop/use-cases/memory-layer/rust/Cargo.toml @@ -22,7 +22,7 @@ lto = "thin" codegen-units = 1 [dependencies] -redis = { version = "0.27", default-features = false } +redis = { version = "1.8", default-features = false } tiny_http = "0.12" serde = { version = "1", features = ["derive"] } serde_json = "1" diff --git a/content/develop/use-cases/memory-layer/rust/src/event_log.rs b/content/develop/use-cases/memory-layer/rust/src/event_log.rs index 73c89bd5ce..61bc9de4b1 100644 --- a/content/develop/use-cases/memory-layer/rust/src/event_log.rs +++ b/content/develop/use-cases/memory-layer/rust/src/event_log.rs @@ -205,7 +205,7 @@ fn redis_value_to_string(v: &Value) -> Option { Value::Int(n) => Some(n.to_string()), Value::Double(d) => Some(d.to_string()), Value::Nil => None, - _ => String::from_redis_value(v).ok(), + _ => String::from_redis_value_ref(v).ok(), } } diff --git a/content/develop/use-cases/memory-layer/rust/src/long_term_memory.rs b/content/develop/use-cases/memory-layer/rust/src/long_term_memory.rs index 3681ab8698..535fd9ba7d 100644 --- a/content/develop/use-cases/memory-layer/rust/src/long_term_memory.rs +++ b/content/develop/use-cases/memory-layer/rust/src/long_term_memory.rs @@ -596,7 +596,7 @@ fn redis_value_to_string(v: &Value) -> Option { Value::Double(d) => Some(d.to_string()), Value::Boolean(b) => Some(b.to_string()), Value::Nil => None, - _ => String::from_redis_value(v).ok(), + _ => String::from_redis_value_ref(v).ok(), } } diff --git a/content/develop/use-cases/memory-layer/rust/src/session_store.rs b/content/develop/use-cases/memory-layer/rust/src/session_store.rs index 80b245a9a6..95962df73d 100644 --- a/content/develop/use-cases/memory-layer/rust/src/session_store.rs +++ b/content/develop/use-cases/memory-layer/rust/src/session_store.rs @@ -345,6 +345,6 @@ fn redis_value_to_string(v: &Value) -> Option { Value::Double(d) => Some(d.to_string()), Value::Boolean(b) => Some(b.to_string()), Value::Nil => None, - _ => String::from_redis_value(v).ok(), + _ => String::from_redis_value_ref(v).ok(), } } diff --git a/content/develop/use-cases/prefetch-cache/rust/Cargo.toml b/content/develop/use-cases/prefetch-cache/rust/Cargo.toml index 0725eb6f4e..1f0b0335a6 100644 --- a/content/develop/use-cases/prefetch-cache/rust/Cargo.toml +++ b/content/develop/use-cases/prefetch-cache/rust/Cargo.toml @@ -4,7 +4,7 @@ version = "0.1.0" edition = "2021" [dependencies] -redis = { version = "0.24", features = ["tokio-comp", "aio", "connection-manager"] } +redis = { version = "1.8", features = ["tokio-comp", "aio", "connection-manager"] } tokio = { version = "1", features = ["full"] } axum = "0.7" serde = { version = "1.0", features = ["derive"] } diff --git a/content/develop/use-cases/prefetch-cache/rust/_index.md b/content/develop/use-cases/prefetch-cache/rust/_index.md index ffccaf5a96..c3704886d1 100644 --- a/content/develop/use-cases/prefetch-cache/rust/_index.md +++ b/content/develop/use-cases/prefetch-cache/rust/_index.md @@ -116,7 +116,7 @@ pub async fn bulk_load( } if loaded > 0 { let mut conn = self.conn.clone(); - pipe.query_async::<_, ()>(&mut conn).await?; + pipe.query_async::<()>(&mut conn).await?; } self.stats.prefetched.fetch_add(loaded, Ordering::Relaxed); Ok(loaded) @@ -172,7 +172,7 @@ pub async fn apply_change(&self, change: &ChangeEvent) -> RedisResult<()> { .del(&cache_key).ignore() .hset_multiple(&cache_key, &pairs).ignore() .expire(&cache_key, self.cfg.ttl_seconds).ignore() - .query_async::<_, ()>(&mut conn).await?; + .query_async::<()>(&mut conn).await?; } "delete" => { let _: i64 = conn.del(&cache_key).await?; @@ -270,7 +270,7 @@ Before running the demo, make sure that: * Redis is running and accessible. By default, the demo connects to `localhost:6379`. * You have a working [Rust toolchain](https://www.rust-lang.org/tools/install) (`cargo` 1.70+). -* The demo's `Cargo.toml` pins the `redis` crate at 0.24+ with the `tokio-comp`, `aio`, and `connection-manager` features; `cargo build` will pull the rest of the dependency graph. +* The demo's `Cargo.toml` pins the `redis` crate at 1.8 with the `tokio-comp`, `aio`, and `connection-manager` features; `cargo build` will pull the rest of the dependency graph. If your Redis server is running elsewhere, start the demo with `--redis-host` and `--redis-port`. @@ -384,7 +384,7 @@ If multiple applications share a Redis deployment, prefix cache keys with the ap ### Use `ConnectionManager`, not a bare connection -`redis::aio::ConnectionManager` is the recommended async connection type for redis-rs. It is cheap to clone (the demo passes a clone to every async task), reconnects automatically on transient failures, and pipelines commands behind the scenes. A bare `redis::aio::Connection` is not `Clone` and cannot be safely shared across tokio tasks. If you need a true connection pool — for example, to fan out heavy `MULTI`/`EXEC` traffic across multiple sockets — use [`deadpool-redis`](https://crates.io/crates/deadpool-redis) or [`bb8-redis`](https://crates.io/crates/bb8-redis). +`redis::aio::ConnectionManager` is the recommended async connection type for redis-rs. It is cheap to clone (the demo passes a clone to every async task), reconnects automatically on transient failures, and pipelines commands behind the scenes. A bare `redis::aio::MultiplexedConnection` is also cheap to clone, but it does not reconnect: once the underlying socket drops, every clone fails until you open a new connection. If you need a true connection pool — for example, to fan out heavy `MULTI`/`EXEC` traffic across multiple sockets — use [`deadpool-redis`](https://crates.io/crates/deadpool-redis) or [`bb8-redis`](https://crates.io/crates/bb8-redis). ### Inspect cached entries directly in Redis diff --git a/content/develop/use-cases/prefetch-cache/rust/cache.rs b/content/develop/use-cases/prefetch-cache/rust/cache.rs index f6f5a4c72e..d93725d56d 100644 --- a/content/develop/use-cases/prefetch-cache/rust/cache.rs +++ b/content/develop/use-cases/prefetch-cache/rust/cache.rs @@ -119,7 +119,7 @@ impl PrefetchCache { } if loaded > 0 { let mut conn = self.conn.clone(); - pipe.query_async::<_, ()>(&mut conn).await?; + pipe.query_async::<()>(&mut conn).await?; } self.stats.prefetched.fetch_add(loaded, Ordering::Relaxed); Ok(loaded) @@ -191,7 +191,7 @@ impl PrefetchCache { .ignore() .expire(&cache_key, self.cfg.ttl_seconds) .ignore() - .query_async::<_, ()>(&mut conn) + .query_async::<()>(&mut conn) .await?; } "delete" => { @@ -230,7 +230,7 @@ impl PrefetchCache { { let mut iter: AsyncIter = conn.scan_match(&pattern).await?; while let Some(key) = iter.next_item().await { - keys.push(key); + keys.push(key?); } } if keys.is_empty() { @@ -256,7 +256,7 @@ impl PrefetchCache { let mut out: Vec = Vec::new(); let mut iter: AsyncIter = conn.scan_match(&pattern).await?; while let Some(key) = iter.next_item().await { - out.push(self.strip_prefix(&key).to_string()); + out.push(self.strip_prefix(&key?).to_string()); } out.sort(); Ok(out) diff --git a/content/develop/use-cases/pub-sub/rust/Cargo.toml b/content/develop/use-cases/pub-sub/rust/Cargo.toml index b464db35e2..e4e8b9972a 100644 --- a/content/develop/use-cases/pub-sub/rust/Cargo.toml +++ b/content/develop/use-cases/pub-sub/rust/Cargo.toml @@ -4,7 +4,7 @@ version = "0.1.0" edition = "2021" [dependencies] -redis = { version = "0.27", features = ["tokio-comp", "aio", "connection-manager"] } +redis = { version = "1.8", features = ["tokio-comp", "aio", "connection-manager"] } tokio = { version = "1", features = ["full"] } axum = "0.7" serde = { version = "1.0", features = ["derive"] } diff --git a/content/develop/use-cases/pub-sub/rust/_index.md b/content/develop/use-cases/pub-sub/rust/_index.md index 6cb17b869d..c6ad65e6df 100644 --- a/content/develop/use-cases/pub-sub/rust/_index.md +++ b/content/develop/use-cases/pub-sub/rust/_index.md @@ -257,7 +257,7 @@ pub async fn stats(&self) -> Stats { * Redis 6.2 or later running locally on the default port (6379). Earlier versions still work for plain `PUBLISH`/`SUBSCRIBE`; `PUBSUB NUMPAT` is older than that. * A Rust toolchain (rustup + cargo) capable of building the 2021 edition. -* The `redis` crate at version 0.27+ with the `tokio-comp`, `aio`, and `connection-manager` features, plus `tokio`, `axum`, `serde`, `serde_json`, and `futures-util` (declared in `Cargo.toml`). +* The `redis` crate at version 1.8 or later with the `tokio-comp`, `aio`, and `connection-manager` features, plus `tokio`, `axum`, `serde`, `serde_json`, and `futures-util` (declared in `Cargo.toml`). ## Running the demo diff --git a/content/develop/use-cases/rate-limiter/rust/Cargo.toml b/content/develop/use-cases/rate-limiter/rust/Cargo.toml index 8c27cbac7a..41b07eed0c 100644 --- a/content/develop/use-cases/rate-limiter/rust/Cargo.toml +++ b/content/develop/use-cases/rate-limiter/rust/Cargo.toml @@ -4,7 +4,7 @@ version = "0.1.0" edition = "2021" [dependencies] -redis = { version = "0.24", features = ["script"] } +redis = { version = "1.8", features = ["script"] } tokio = { version = "1", features = ["full"] } axum = "0.7" serde = { version = "1.0", features = ["derive"] } diff --git a/content/develop/use-cases/rate-limiter/rust/_index.md b/content/develop/use-cases/rate-limiter/rust/_index.md index 58d54fa1bc..c5b331e089 100644 --- a/content/develop/use-cases/rate-limiter/rust/_index.md +++ b/content/develop/use-cases/rate-limiter/rust/_index.md @@ -121,14 +121,14 @@ Add the `redis` crate to your `Cargo.toml`: ```toml [dependencies] -redis = "0.24" +redis = "1.8" ``` For async support, enable the `tokio-comp` or `async-std-comp` feature: ```toml [dependencies] -redis = { version = "0.24", features = ["tokio-comp"] } +redis = { version = "1.8", features = ["tokio-comp"] } tokio = { version = "1", features = ["full"] } ``` @@ -328,11 +328,11 @@ fn handle_request( For async web frameworks like `actix-web` or `axum`, use the async version of the Redis client: ```rust -use redis::aio::Connection; +use redis::aio::MultiplexedConnection; async fn rate_limit_check( limiter: &TokenBucket, - con: &mut Connection, + con: &mut MultiplexedConnection, key: &str, ) -> redis::RedisResult { limiter.allow_async(con, key).await diff --git a/content/develop/use-cases/rate-limiter/rust/token_bucket.rs b/content/develop/use-cases/rate-limiter/rust/token_bucket.rs index 54eca45ec5..3bc17614c3 100644 --- a/content/develop/use-cases/rate-limiter/rust/token_bucket.rs +++ b/content/develop/use-cases/rate-limiter/rust/token_bucket.rs @@ -171,7 +171,7 @@ impl TokenBucket { let remaining = match result.get(1) { Some(redis::Value::Int(v)) => *v as f64, - Some(redis::Value::Data(v)) => { + Some(redis::Value::BulkString(v)) => { String::from_utf8_lossy(v).parse::().unwrap_or(0.0) } _ => 0.0, diff --git a/content/develop/use-cases/recommendation-engine/rust/Cargo.toml b/content/develop/use-cases/recommendation-engine/rust/Cargo.toml index d674a3813a..3654ee2366 100644 --- a/content/develop/use-cases/recommendation-engine/rust/Cargo.toml +++ b/content/develop/use-cases/recommendation-engine/rust/Cargo.toml @@ -4,7 +4,7 @@ version = "0.1.0" edition = "2021" [dependencies] -redis = { version = "0.27", features = ["tokio-comp", "aio", "connection-manager"] } +redis = { version = "1.8", features = ["tokio-comp", "aio", "connection-manager"] } tokio = { version = "1", features = ["full"] } axum = "0.7" serde = { version = "1.0", features = ["derive"] } diff --git a/content/develop/use-cases/recommendation-engine/rust/recommender.rs b/content/develop/use-cases/recommendation-engine/rust/recommender.rs index 1c56165f53..806b734b87 100644 --- a/content/develop/use-cases/recommendation-engine/rust/recommender.rs +++ b/content/develop/use-cases/recommendation-engine/rust/recommender.rs @@ -250,7 +250,7 @@ impl RedisRecommender { for p in products { let key = self.product_key(&p.id); let vec_bytes = decode_embedding(p).map_err(|e| { - RedisError::from((redis::ErrorKind::TypeError, "encode product", e)) + RedisError::from((redis::ErrorKind::UnexpectedReturnType, "encode product", e)) })?; // Build the field list explicitly so we can mix string and // byte values in the same HSET. Strings go through as @@ -412,20 +412,20 @@ impl RedisRecommender { // `HMGET` returns `[Option, Option]` here because the // embedding is binary. let fields: Vec>> = conn - .hget(&product_key, &["embedding", "category"]) + .hmget(&product_key, &["embedding", "category"]) .await?; let clicked_bytes = match fields.get(0).and_then(|v| v.as_ref()) { Some(b) => b.clone(), None => { return Err(RedisError::from(( - redis::ErrorKind::TypeError, + redis::ErrorKind::UnexpectedReturnType, "unknown product", format!("no hash at {}", product_key), ))) } }; let clicked_vec = bytes_to_floats(&clicked_bytes).map_err(|e| { - RedisError::from((redis::ErrorKind::TypeError, "decode embedding", e)) + RedisError::from((redis::ErrorKind::UnexpectedReturnType, "decode embedding", e)) })?; let category = fields .get(1) @@ -439,7 +439,7 @@ impl RedisRecommender { let new_session = match previous { Some(prev_bytes) if !prev_bytes.is_empty() => { let prev_vec = bytes_to_floats(&prev_bytes).map_err(|e| { - RedisError::from((redis::ErrorKind::TypeError, "decode session_vec", e)) + RedisError::from((redis::ErrorKind::UnexpectedReturnType, "decode session_vec", e)) })?; ewma_blend(&prev_vec, &clicked_vec, ewma_alpha) } @@ -494,7 +494,7 @@ impl RedisRecommender { "session_vec" => { let vec = bytes_to_floats(&v).map_err(|e| { RedisError::from(( - redis::ErrorKind::TypeError, + redis::ErrorKind::UnexpectedReturnType, "decode session_vec", e, )) @@ -541,7 +541,7 @@ impl RedisRecommender { /// The HNSW index reflects the change as soon as the `HSET` /// commits, so subsequent `FT.SEARCH` calls see the new vector /// without any index rebuild or serving downtime. Returns - /// `ErrorKind::TypeError` (with detail "unknown product") if the + /// `ErrorKind::UnexpectedReturnType` (with detail "unknown product") if the /// product hash does not exist (HSET would otherwise happily /// create a partial document that the index then picks up). Also /// rejects vectors with the wrong dimensionality so a model swap @@ -553,7 +553,7 @@ impl RedisRecommender { ) -> RedisResult<()> { if new_vector.len() != self.vector_dim { return Err(RedisError::from(( - redis::ErrorKind::TypeError, + redis::ErrorKind::UnexpectedReturnType, "bad vector dim", format!( "new_vector has length {}; index expects {}", @@ -567,7 +567,7 @@ impl RedisRecommender { let exists: i64 = conn.exists(&key).await?; if exists == 0 { return Err(RedisError::from(( - redis::ErrorKind::TypeError, + redis::ErrorKind::UnexpectedReturnType, "unknown product", format!("no hash at {}", key), ))); @@ -815,20 +815,20 @@ fn parse_search_reply(raw: Value, key_prefix: &str) -> RedisResult items, _ => { return Err(RedisError::from(( - redis::ErrorKind::TypeError, + redis::ErrorKind::UnexpectedReturnType, "FT.SEARCH: expected array reply", ))); } }; let mut iter = items.into_iter(); let _total: i64 = match iter.next() { - Some(v) => redis::FromRedisValue::from_redis_value(&v).unwrap_or(0), + Some(v) => redis::FromRedisValue::from_redis_value_ref(&v).unwrap_or(0), None => 0, }; let mut out: Vec = Vec::new(); while let Some(key_v) = iter.next() { let key: String = - redis::FromRedisValue::from_redis_value(&key_v).unwrap_or_default(); + redis::FromRedisValue::from_redis_value_ref(&key_v).unwrap_or_default(); let Some(fields_v) = iter.next() else { break }; let fields = match fields_v { Value::Array(items) => items, @@ -838,9 +838,9 @@ fn parse_search_reply(raw: Value, key_prefix: &str) -> RedisResult Option> { let mut out: HashMap = HashMap::new(); let mut iter = items.iter(); while let (Some(k), Some(value)) = (iter.next(), iter.next()) { - let kk: String = match redis::FromRedisValue::from_redis_value(k) { + let kk: String = match redis::FromRedisValue::from_redis_value_ref(k) { Ok(s) => s, Err(_) => continue, }; @@ -894,18 +894,16 @@ fn flat_array_to_map(v: &Value) -> Option> { } fn value_as_i64(v: &Value) -> Option { - redis::FromRedisValue::from_redis_value(v).ok() + redis::FromRedisValue::from_redis_value_ref(v).ok() } fn value_as_f64(v: &Value) -> Option { // FT.INFO returns size-in-mb as a string in many versions; tolerate // both forms. - let as_f: RedisResult = redis::FromRedisValue::from_redis_value(v); - if let Ok(n) = as_f { + if let Ok(n) = ::from_redis_value_ref(v) { return Some(n); } - let as_s: RedisResult = redis::FromRedisValue::from_redis_value(v); - if let Ok(s) = as_s { + if let Ok(s) = ::from_redis_value_ref(v) { if let Ok(n) = s.parse::() { return Some(n); } diff --git a/content/develop/use-cases/semantic-cache/rust/Cargo.lock b/content/develop/use-cases/semantic-cache/rust/Cargo.lock index 194c45868f..64ceca2801 100644 --- a/content/develop/use-cases/semantic-cache/rust/Cargo.lock +++ b/content/develop/use-cases/semantic-cache/rust/Cargo.lock @@ -33,13 +33,10 @@ dependencies = [ ] [[package]] -name = "arc-swap" -version = "1.9.1" +name = "arcstr" +version = "1.2.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "6a3a1fd6f75306b68087b831f025c712524bcb19aad54e557b1129cfa0a2b207" -dependencies = [ - "rustversion", -] +checksum = "03918c3dbd7701a85c6b9887732e2921175f26c350b4563841d0958c21d57e6d" [[package]] name = "ascii" @@ -210,9 +207,9 @@ checksum = "6e4de3bc4ea267985becf712dc6d9eed8b04c953b3fcfb339ebc87acd9804901" [[package]] name = "combine" -version = "4.6.7" +version = "4.6.9" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "ba5a308b75df32fe02788e748662718f03fde005016435c444eea572398219fd" +checksum = "4356eba6976435498d05a311ee9fc486da3bc81c200d3f0d7dd9703b41913f92" dependencies = [ "bytes", "memchr", @@ -1057,15 +1054,6 @@ dependencies = [ "either", ] -[[package]] -name = "itertools" -version = "0.13.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "413ee7dfc52ee1a4949ceeb7dbc8a33f2d6c088194d9f922fb8318faf1f01186" -dependencies = [ - "either", -] - [[package]] name = "itoa" version = "1.0.18" @@ -1681,18 +1669,18 @@ checksum = "03251193000f4bd3b042892be858ee50e8b3719f2b08e5833ac4353724632430" [[package]] name = "redis" -version = "0.27.6" +version = "1.8.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "09d8f99a4090c89cc489a94833c901ead69bfbf3877b4867d5482e321ee875bc" +checksum = "7acd809e80afa754c63f586955699d8990efe1c4b6613183ee5f027a11fbc8b5" dependencies = [ - "arc-swap", + "arcstr", "combine", - "itertools 0.13.0", "itoa", - "num-bigint", "percent-encoding", "ryu", + "socket2", "url", + "xxhash-rust", ] [[package]] @@ -1966,6 +1954,16 @@ version = "1.15.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "67b1b7a3b5fe4f1376887184045fcf45c69e92af734b7aaddc05fb777b6fbd03" +[[package]] +name = "socket2" +version = "0.6.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c3d1e2c7f27f8d4cb10542a02c49005dbd6e93095799d6f3be745fae9f8fedd4" +dependencies = [ + "libc", + "windows-sys 0.61.2", +] + [[package]] name = "spm_precompiled" version = "0.1.4" @@ -2730,6 +2728,12 @@ version = "0.6.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "1ffae5123b2d3fc086436f8834ae3ab053a283cfac8fe0a0b8eaae044768a4c4" +[[package]] +name = "xxhash-rust" +version = "0.8.19" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "550a2b930b62486a393c52d5c3b84bff264b28aa437ed64694d31e93b1757af7" + [[package]] name = "yoke" version = "0.7.5" diff --git a/content/develop/use-cases/semantic-cache/rust/Cargo.toml b/content/develop/use-cases/semantic-cache/rust/Cargo.toml index d084be3cd6..86d591c696 100644 --- a/content/develop/use-cases/semantic-cache/rust/Cargo.toml +++ b/content/develop/use-cases/semantic-cache/rust/Cargo.toml @@ -18,7 +18,7 @@ lto = "thin" codegen-units = 1 [dependencies] -redis = { version = "0.27", default-features = false } +redis = { version = "1.8", default-features = false } tiny_http = "0.12" serde = { version = "1", features = ["derive"] } serde_json = "1" diff --git a/content/develop/use-cases/semantic-cache/rust/_index.md b/content/develop/use-cases/semantic-cache/rust/_index.md index 313ab36a8c..0f77943167 100644 --- a/content/develop/use-cases/semantic-cache/rust/_index.md +++ b/content/develop/use-cases/semantic-cache/rust/_index.md @@ -237,7 +237,7 @@ The server holds one `LocalEmbedder`, one `RedisSemanticCache`, and one `MockLlm cd docs/content/develop/use-cases/semantic-cache/rust ``` -2. Make sure a recent Rust toolchain is installed (Rust 1.75 or newer). +2. Make sure a recent Rust toolchain is installed (Rust 1.88 or newer). [`rustup`](https://rustup.rs/) is the usual route. The build pulls in Candle and the Hugging Face tokenizer crates, so the first `cargo build` takes a few minutes. diff --git a/content/develop/use-cases/semantic-cache/rust/src/cache.rs b/content/develop/use-cases/semantic-cache/rust/src/cache.rs index 79af117090..714be823c7 100644 --- a/content/develop/use-cases/semantic-cache/rust/src/cache.rs +++ b/content/develop/use-cases/semantic-cache/rust/src/cache.rs @@ -667,7 +667,7 @@ fn redis_value_to_string(v: &Value) -> Option { // variants we didn't list explicitly (Map, Set, BigNumber, // etc — these shouldn't appear in FT.SEARCH replies but // future protocol additions could). - String::from_redis_value(v).ok() + String::from_redis_value_ref(v).ok() } } } diff --git a/content/develop/use-cases/session-store/rust/Cargo.toml b/content/develop/use-cases/session-store/rust/Cargo.toml index 5287487470..4be9adec62 100644 --- a/content/develop/use-cases/session-store/rust/Cargo.toml +++ b/content/develop/use-cases/session-store/rust/Cargo.toml @@ -4,7 +4,7 @@ version = "0.1.0" edition = "2021" [dependencies] -redis = { version = "0.24", features = ["tokio-comp"] } +redis = { version = "1.8", features = ["tokio-comp"] } tokio = { version = "1", features = ["full"] } axum = "0.7" serde = { version = "1.0", features = ["derive"] } diff --git a/content/develop/use-cases/session-store/rust/_index.md b/content/develop/use-cases/session-store/rust/_index.md index 1e08d9bcc5..683bb43dd0 100644 --- a/content/develop/use-cases/session-store/rust/_index.md +++ b/content/develop/use-cases/session-store/rust/_index.md @@ -88,7 +88,7 @@ use std::collections::HashMap; #[tokio::main] async fn main() -> redis::RedisResult<()> { let client = Client::open("redis://localhost:6379/")?; - let mut con = client.get_async_connection().await?; + let mut con = client.get_multiplexed_async_connection().await?; let store = RedisSessionStore::new("session:", 1800)?; @@ -176,7 +176,7 @@ pub async fn get_session_async( refresh_ttl: bool, ) -> RedisResult>> where - C: redis::aio::ConnectionLike + Send, + C: redis::aio::ConnectionLike + Send + Sync, { let key = self.session_key(session_id); let session: HashMap = con.hgetall(&key).await?; @@ -205,7 +205,7 @@ Add the crates you need to `Cargo.toml`: ```toml [dependencies] -redis = { version = "0.24", features = ["tokio-comp"] } +redis = { version = "1.8", features = ["tokio-comp"] } tokio = { version = "1", features = ["full"] } axum = "0.7" serde = { version = "1.0", features = ["derive"] } diff --git a/content/develop/use-cases/session-store/rust/demo_server.rs b/content/develop/use-cases/session-store/rust/demo_server.rs index 1e6cadad96..83a436a812 100644 --- a/content/develop/use-cases/session-store/rust/demo_server.rs +++ b/content/develop/use-cases/session-store/rust/demo_server.rs @@ -15,7 +15,7 @@ use axum::{ routing::{get, post}, Json, Router, }; -use redis::{aio::Connection, Client}; +use redis::{aio::MultiplexedConnection, Client}; use serde::Serialize; use session_store::RedisSessionStore; use std::{collections::HashMap, env, sync::Arc}; @@ -23,7 +23,7 @@ use tokio::sync::Mutex; #[derive(Clone)] struct AppState { - redis_connection: Arc>, + redis_connection: Arc>, store: RedisSessionStore, } @@ -55,7 +55,7 @@ async fn main() { std::process::exit(1); }); - let connection = client.get_async_connection().await.unwrap_or_else(|e| { + let connection = client.get_multiplexed_async_connection().await.unwrap_or_else(|e| { eprintln!("Failed to connect to Redis at {}: {}", redis_url, e); std::process::exit(1); }); diff --git a/content/develop/use-cases/session-store/rust/session_store.rs b/content/develop/use-cases/session-store/rust/session_store.rs index b713656b30..0a8e8c22c1 100644 --- a/content/develop/use-cases/session-store/rust/session_store.rs +++ b/content/develop/use-cases/session-store/rust/session_store.rs @@ -64,7 +64,7 @@ impl RedisSessionStore { ttl: Option, ) -> RedisResult where - C: redis::aio::ConnectionLike + Send, + C: redis::aio::ConnectionLike + Send + Sync, { let session_id = self.create_session_id(); let key = self.session_key(&session_id); @@ -101,7 +101,7 @@ impl RedisSessionStore { session_id: &str, ) -> RedisResult> where - C: redis::aio::ConnectionLike + Send, + C: redis::aio::ConnectionLike + Send + Sync, { let stored_ttl: Option = con.hget(self.session_key(session_id), "session_ttl").await?; match stored_ttl { @@ -146,7 +146,7 @@ impl RedisSessionStore { refresh_ttl: bool, ) -> RedisResult>> where - C: redis::aio::ConnectionLike + Send, + C: redis::aio::ConnectionLike + Send + Sync, { let key = self.session_key(session_id); let session: HashMap = con.hgetall(&key).await?; @@ -210,7 +210,7 @@ impl RedisSessionStore { data: &HashMap, ) -> RedisResult where - C: redis::aio::ConnectionLike + Send, + C: redis::aio::ConnectionLike + Send + Sync, { let key = self.session_key(session_id); let session: HashMap = con.hgetall(&key).await?; @@ -269,7 +269,7 @@ impl RedisSessionStore { amount: i64, ) -> RedisResult> where - C: redis::aio::ConnectionLike + Send, + C: redis::aio::ConnectionLike + Send + Sync, { let key = self.session_key(session_id); let session: HashMap = con.hgetall(&key).await?; @@ -319,7 +319,7 @@ impl RedisSessionStore { ttl: usize, ) -> RedisResult where - C: redis::aio::ConnectionLike + Send, + C: redis::aio::ConnectionLike + Send + Sync, { let key = self.session_key(session_id); let session: HashMap = con.hgetall(&key).await?; @@ -355,7 +355,7 @@ impl RedisSessionStore { session_id: &str, ) -> RedisResult where - C: redis::aio::ConnectionLike + Send, + C: redis::aio::ConnectionLike + Send + Sync, { let deleted: i64 = con.del(self.session_key(session_id)).await?; Ok(deleted == 1) @@ -378,7 +378,7 @@ impl RedisSessionStore { session_id: &str, ) -> RedisResult where - C: redis::aio::ConnectionLike + Send, + C: redis::aio::ConnectionLike + Send + Sync, { con.ttl(self.session_key(session_id)).await } @@ -389,7 +389,7 @@ impl RedisSessionStore { fn normalize_ttl_value(ttl: usize) -> RedisResult { if ttl < 1 { - Err(RedisError::from((ErrorKind::TypeError, "TTL must be at least 1 second"))) + Err(RedisError::from((ErrorKind::UnexpectedReturnType, "TTL must be at least 1 second"))) } else { Ok(ttl) } @@ -402,11 +402,11 @@ impl RedisSessionStore { fn timestamp(&self) -> RedisResult { let mut ts = OffsetDateTime::now_utc() .replace_nanosecond(0) - .map_err(|_| RedisError::from((ErrorKind::TypeError, "invalid timestamp")))?; + .map_err(|_| RedisError::from((ErrorKind::UnexpectedReturnType, "invalid timestamp")))?; ts = ts.to_offset(time::UtcOffset::UTC); let formatted = ts .format(&Rfc3339) - .map_err(|_| RedisError::from((ErrorKind::TypeError, "invalid timestamp format")))?; + .map_err(|_| RedisError::from((ErrorKind::UnexpectedReturnType, "invalid timestamp format")))?; Ok(formatted.replace('Z', "+00:00")) } @@ -451,8 +451,8 @@ impl RedisSessionStore { fn parse_ttl(session: &HashMap) -> RedisResult { session .get("session_ttl") - .ok_or_else(|| RedisError::from((ErrorKind::TypeError, "missing session_ttl")))? + .ok_or_else(|| RedisError::from((ErrorKind::UnexpectedReturnType, "missing session_ttl")))? .parse::() - .map_err(|_| RedisError::from((ErrorKind::TypeError, "invalid session_ttl"))) + .map_err(|_| RedisError::from((ErrorKind::UnexpectedReturnType, "invalid session_ttl"))) } } diff --git a/content/develop/use-cases/streaming/rust/Cargo.toml b/content/develop/use-cases/streaming/rust/Cargo.toml index 6b27036b79..f81c9bf1ee 100644 --- a/content/develop/use-cases/streaming/rust/Cargo.toml +++ b/content/develop/use-cases/streaming/rust/Cargo.toml @@ -4,7 +4,7 @@ version = "0.1.0" edition = "2021" [dependencies] -redis = { version = "0.24", features = ["tokio-comp", "aio", "connection-manager", "streams"] } +redis = { version = "1.8", features = ["tokio-comp", "aio", "connection-manager", "streams"] } tokio = { version = "1", features = ["full"] } axum = "0.7" serde = { version = "1.0", features = ["derive"] } diff --git a/content/develop/use-cases/streaming/rust/_index.md b/content/develop/use-cases/streaming/rust/_index.md index e532820594..3e22150844 100644 --- a/content/develop/use-cases/streaming/rust/_index.md +++ b/content/develop/use-cases/streaming/rust/_index.md @@ -52,12 +52,15 @@ The `EventStream` struct wraps the stream operations ([source](https://github.com/redis/docs/blob/main/content/develop/use-cases/streaming/rust/event_stream.rs)): ```rust -use redis::aio::ConnectionManager; +use redis::aio::{ConnectionManager, ConnectionManagerConfig}; use redis::Client; use std::collections::HashMap; let client = Client::open("redis://127.0.0.1:6379/")?; -let conn = ConnectionManager::new(client).await?; +// Consumers block in XREADGROUP ... BLOCK for longer than the 500 ms default +// response timeout in redis-rs 1.x, so turn the response timeout off. +let config = ConnectionManagerConfig::new().set_response_timeout(None); +let conn = ConnectionManager::new_with_config(client, config).await?; let stream = EventStream::new( conn, "demo:events:orders", // stream key @@ -252,7 +255,7 @@ pub async fn reap_idle_pel(&self) -> ReapResult { } ``` -The underlying `stream.autoclaim` helper pages through the group's PEL with `XAUTOCLAIM`'s continuation cursor. redis-rs 0.24 has no typed wrapper for `XAUTOCLAIM`, so the helper builds it from raw `redis::cmd("XAUTOCLAIM")` and decodes the three-element Redis 7+ reply by hand: +The underlying `stream.autoclaim` helper pages through the group's PEL with `XAUTOCLAIM`'s continuation cursor. the helper builds it from the generic `redis::cmd("XAUTOCLAIM")` builder (redis-rs 1.x also has a typed `xautoclaim_options`) and decodes the three-element Redis 7+ reply by hand: ```rust pub async fn autoclaim( @@ -372,14 +375,14 @@ The pause and crash levers exist only for the demo. A real consumer is just the * Redis 7.0 or later. `XAUTOCLAIM` was added in Redis 6.2, but its reply gained a third element (the list of deleted IDs) in 7.0; the example relies on that shape. -* Rust 1.70 or later (stable). -* The `redis` crate at version 0.24+, with features `["tokio-comp", "aio", "connection-manager", "streams"]`. The `connection-manager` feature gives `ConnectionManager` for cheap, cloneable, auto-reconnecting connections; `streams` adds the typed reply structs for `XINFO`/`XPENDING`. +* Rust 1.88 or later (the `redis` 1.8 crate requires it). +* The `redis` crate at version 1.8 or later, with features `["tokio-comp", "aio", "connection-manager", "streams"]`. The `connection-manager` feature gives `ConnectionManager` for cheap, cloneable, auto-reconnecting connections; `streams` adds the typed reply structs for `XINFO`/`XPENDING`. The `Cargo.toml` for this demo pins all the runtime crates: ```toml [dependencies] -redis = { version = "0.24", features = ["tokio-comp", "aio", "connection-manager", "streams"] } +redis = { version = "1.8", features = ["tokio-comp", "aio", "connection-manager", "streams"] } tokio = { version = "1", features = ["full"] } axum = "0.7" serde = { version = "1.0", features = ["derive"] } diff --git a/content/develop/use-cases/streaming/rust/demo_server.rs b/content/develop/use-cases/streaming/rust/demo_server.rs index 9dee2c6580..a60d5e8c4f 100644 --- a/content/develop/use-cases/streaming/rust/demo_server.rs +++ b/content/develop/use-cases/streaming/rust/demo_server.rs @@ -345,7 +345,10 @@ async fn main() { let url = format!("redis://{}:{}/", redis_host, redis_port); let client = Client::open(url).expect("failed to create Redis client"); - let conn = ConnectionManager::new(client) + // Consumers block in XREADGROUP ... BLOCK for longer than the 500 ms default + // response timeout in redis-rs 1.x, so turn the response timeout off. + let config = redis::aio::ConnectionManagerConfig::new().set_response_timeout(None); + let conn = ConnectionManager::new_with_config(client, config) .await .expect("failed to connect to Redis"); diff --git a/content/develop/use-cases/streaming/rust/event_stream.rs b/content/develop/use-cases/streaming/rust/event_stream.rs index 77f55ccada..6e076f5ea7 100644 --- a/content/develop/use-cases/streaming/rust/event_stream.rs +++ b/content/develop/use-cases/streaming/rust/event_stream.rs @@ -461,17 +461,17 @@ impl EventStream { Ok(v) => v, Err(_) => return StreamInfo::default(), }; - let map: HashMap = match redis::FromRedisValue::from_redis_value(&raw) { + let map: HashMap = match redis::FromRedisValue::from_redis_value_ref(&raw) { Ok(m) => m, Err(_) => return StreamInfo::default(), }; let length: u64 = map .get("length") - .and_then(|v| redis::FromRedisValue::from_redis_value(v).ok()) + .and_then(|v| redis::FromRedisValue::from_redis_value_ref(v).ok()) .unwrap_or(0); let last_generated_id: Option = map .get("last-generated-id") - .and_then(|v| redis::FromRedisValue::from_redis_value(v).ok()); + .and_then(|v| redis::FromRedisValue::from_redis_value_ref(v).ok()); let first_entry_id = entry_id_from_value(map.get("first-entry")); let last_entry_id = entry_id_from_value(map.get("last-entry")); StreamInfo { @@ -498,7 +498,7 @@ impl EventStream { // a vec of HashMap so we can pull only the keys // we care about (and tolerate version differences). let rows: Vec> = - match redis::FromRedisValue::from_redis_value(&raw) { + match redis::FromRedisValue::from_redis_value_ref(&raw) { Ok(v) => v, Err(_) => return Vec::new(), }; @@ -506,23 +506,23 @@ impl EventStream { .map(|map| GroupInfo { name: map .get("name") - .and_then(|v| redis::FromRedisValue::from_redis_value(v).ok()) + .and_then(|v| redis::FromRedisValue::from_redis_value_ref(v).ok()) .unwrap_or_default(), consumers: map .get("consumers") - .and_then(|v| redis::FromRedisValue::from_redis_value(v).ok()) + .and_then(|v| redis::FromRedisValue::from_redis_value_ref(v).ok()) .unwrap_or(0), pending: map .get("pending") - .and_then(|v| redis::FromRedisValue::from_redis_value(v).ok()) + .and_then(|v| redis::FromRedisValue::from_redis_value_ref(v).ok()) .unwrap_or(0), last_delivered_id: map .get("last-delivered-id") - .and_then(|v| redis::FromRedisValue::from_redis_value(v).ok()) + .and_then(|v| redis::FromRedisValue::from_redis_value_ref(v).ok()) .unwrap_or_default(), lag: map .get("lag") - .and_then(|v| redis::FromRedisValue::from_redis_value(v).ok()), + .and_then(|v| redis::FromRedisValue::from_redis_value_ref(v).ok()), }) .collect() } @@ -541,7 +541,7 @@ impl EventStream { Err(_) => return Vec::new(), }; let rows: Vec> = - match redis::FromRedisValue::from_redis_value(&raw) { + match redis::FromRedisValue::from_redis_value_ref(&raw) { Ok(v) => v, Err(_) => return Vec::new(), }; @@ -549,15 +549,15 @@ impl EventStream { .map(|map| ConsumerInfo { name: map .get("name") - .and_then(|v| redis::FromRedisValue::from_redis_value(v).ok()) + .and_then(|v| redis::FromRedisValue::from_redis_value_ref(v).ok()) .unwrap_or_default(), pending: map .get("pending") - .and_then(|v| redis::FromRedisValue::from_redis_value(v).ok()) + .and_then(|v| redis::FromRedisValue::from_redis_value_ref(v).ok()) .unwrap_or(0), idle_ms: map .get("idle") - .and_then(|v| redis::FromRedisValue::from_redis_value(v).ok()) + .and_then(|v| redis::FromRedisValue::from_redis_value_ref(v).ok()) .unwrap_or(0), }) .collect() @@ -675,7 +675,7 @@ fn stream_ids_to_entries(ids: Vec) -> Vec { fn fields_from_stream_map(map: &HashMap) -> HashMap { let mut out: HashMap = HashMap::with_capacity(map.len()); for (k, v) in map { - let s: String = redis::FromRedisValue::from_redis_value(v).unwrap_or_default(); + let s: String = redis::FromRedisValue::from_redis_value_ref(v).unwrap_or_default(); out.insert(k.clone(), s); } out @@ -688,9 +688,9 @@ fn entry_id_from_value(v: Option<&Value>) -> Option { if matches!(v, Value::Nil) { return None; } - if let Value::Bulk(items) = v { + if let Value::Array(items) = v { if let Some(first) = items.first() { - if let Ok(id) = redis::FromRedisValue::from_redis_value(first) { + if let Ok(id) = redis::FromRedisValue::from_redis_value_ref(first) { return Some(id); } } @@ -702,10 +702,10 @@ fn entry_id_from_value(v: Option<&Value>) -> Option { /// `[ next-cursor, [ [id, [field, value, ...]], ... ], [deleted-id, ...] ]`. fn parse_autoclaim_reply(raw: Value) -> RedisResult<(String, Vec, Vec)> { let parts = match raw { - Value::Bulk(parts) => parts, + Value::Array(parts) => parts, _ => { return Err(RedisError::from(( - redis::ErrorKind::TypeError, + redis::ErrorKind::UnexpectedReturnType, "XAUTOCLAIM: expected array reply", ))) } @@ -713,22 +713,22 @@ fn parse_autoclaim_reply(raw: Value) -> RedisResult<(String, Vec, Vec = match deleted_v { - Some(v) => redis::FromRedisValue::from_redis_value(&v).unwrap_or_default(), + Some(v) => redis::FromRedisValue::from_redis_value_ref(&v).unwrap_or_default(), None => Vec::new(), }; diff --git a/content/develop/use-cases/time-series-dashboard/rust/Cargo.toml b/content/develop/use-cases/time-series-dashboard/rust/Cargo.toml index 88cfb1b401..dec79d11f6 100644 --- a/content/develop/use-cases/time-series-dashboard/rust/Cargo.toml +++ b/content/develop/use-cases/time-series-dashboard/rust/Cargo.toml @@ -6,7 +6,7 @@ edition = "2021" [dependencies] axum = "0.7" rand = "0.8" -redis = { version = "0.24", features = ["tokio-comp"] } +redis = { version = "1.8", features = ["tokio-comp"] } serde = { version = "1.0", features = ["derive"] } serde_json = "1.0" tokio = { version = "1", features = ["full"] } diff --git a/content/develop/use-cases/time-series-dashboard/rust/demo_server.rs b/content/develop/use-cases/time-series-dashboard/rust/demo_server.rs index 48414b58d1..37119e246f 100644 --- a/content/develop/use-cases/time-series-dashboard/rust/demo_server.rs +++ b/content/develop/use-cases/time-series-dashboard/rust/demo_server.rs @@ -13,7 +13,7 @@ use axum::{ routing::get, Json, Router, }; -use redis::{aio::Connection, Client}; +use redis::{aio::MultiplexedConnection, Client}; use sensor_simulator::{sensors, SensorSimulator}; use std::{env, sync::Arc, time::Duration}; use tokio::sync::Mutex; @@ -28,7 +28,7 @@ const GRAPH_PAD_Y: i32 = 18; #[derive(Clone)] struct AppState { - redis_connection: Arc>, + redis_connection: Arc>, store: RedisTimeSeriesStore, } @@ -46,7 +46,7 @@ async fn main() { std::process::exit(1); }); - let connection = client.get_async_connection().await.unwrap_or_else(|error| { + let connection = client.get_multiplexed_async_connection().await.unwrap_or_else(|error| { eprintln!("Failed to connect to Redis at {}: {}", redis_url, error); std::process::exit(1); }); diff --git a/content/develop/use-cases/time-series-dashboard/rust/timeseries_store.rs b/content/develop/use-cases/time-series-dashboard/rust/timeseries_store.rs index 6d57a0f57d..2c44cc5e8b 100644 --- a/content/develop/use-cases/time-series-dashboard/rust/timeseries_store.rs +++ b/content/develop/use-cases/time-series-dashboard/rust/timeseries_store.rs @@ -226,7 +226,7 @@ impl RedisTimeSeriesStore { }; match value { - Value::Bulk(values) if values.len() >= 2 => Ok(Some(LatestValue { + Value::Array(values) if values.len() >= 2 => Ok(Some(LatestValue { timestamp: as_i64(&values[0])?, value: as_f64(&values[1])?, })), @@ -239,19 +239,19 @@ impl RedisTimeSeriesStore { fn now_millis() -> RedisResult { let duration = SystemTime::now() .duration_since(UNIX_EPOCH) - .map_err(|_| RedisError::from((ErrorKind::TypeError, "system clock went backwards")))?; + .map_err(|_| RedisError::from((ErrorKind::UnexpectedReturnType, "system clock went backwards")))?; i64::try_from(duration.as_millis()) - .map_err(|_| RedisError::from((ErrorKind::TypeError, "timestamp overflow"))) + .map_err(|_| RedisError::from((ErrorKind::UnexpectedReturnType, "timestamp overflow"))) } fn parse_points(value: Value) -> RedisResult> { let rows = match value { - Value::Bulk(rows) => rows, + Value::Array(rows) => rows, Value::Nil => return Ok(Vec::new()), other => { return Err(RedisError::from(( - ErrorKind::TypeError, + ErrorKind::UnexpectedReturnType, "unexpected TS.RANGE response", format!("{other:?}"), ))) @@ -260,7 +260,7 @@ fn parse_points(value: Value) -> RedisResult> { let mut points = Vec::with_capacity(rows.len()); for row in rows { - let Value::Bulk(pair) = row else { + let Value::Array(pair) = row else { continue; }; if pair.len() < 2 { @@ -279,11 +279,11 @@ fn parse_points(value: Value) -> RedisResult> { fn as_i64(value: &Value) -> RedisResult { match value { Value::Int(number) => Ok(*number), - Value::Data(bytes) => String::from_utf8_lossy(bytes) + Value::BulkString(bytes) => String::from_utf8_lossy(bytes) .parse::() - .map_err(|_| RedisError::from((ErrorKind::TypeError, "expected integer value"))), + .map_err(|_| RedisError::from((ErrorKind::UnexpectedReturnType, "expected integer value"))), other => Err(RedisError::from(( - ErrorKind::TypeError, + ErrorKind::UnexpectedReturnType, "expected integer value", format!("{other:?}"), ))), @@ -293,11 +293,16 @@ fn as_i64(value: &Value) -> RedisResult { fn as_f64(value: &Value) -> RedisResult { match value { Value::Int(number) => Ok(*number as f64), - Value::Data(bytes) => String::from_utf8_lossy(bytes) + Value::Double(number) => Ok(*number), + // Newer RedisTimeSeries replies send RESP2 doubles as simple strings. + Value::SimpleString(text) => text .parse::() - .map_err(|_| RedisError::from((ErrorKind::TypeError, "expected floating-point value"))), + .map_err(|_| RedisError::from((ErrorKind::UnexpectedReturnType, "expected floating-point value"))), + Value::BulkString(bytes) => String::from_utf8_lossy(bytes) + .parse::() + .map_err(|_| RedisError::from((ErrorKind::UnexpectedReturnType, "expected floating-point value"))), other => Err(RedisError::from(( - ErrorKind::TypeError, + ErrorKind::UnexpectedReturnType, "expected floating-point value", format!("{other:?}"), ))),