redis_oxide/
transaction.rs

1//! Transaction support for Redis
2//!
3//! This module provides functionality for Redis transactions using MULTI/EXEC/WATCH/DISCARD.
4//! Redis transactions allow you to execute a group of commands atomically.
5//!
6//! # Examples
7//!
8//! ```no_run
9//! use redis_oxide::{Client, ConnectionConfig};
10//!
11//! # #[tokio::main]
12//! # async fn main() -> Result<(), Box<dyn std::error::Error>> {
13//! let config = ConnectionConfig::new("redis://localhost:6379");
14//! let client = Client::connect(config).await?;
15//!
16//! // Start a transaction
17//! let mut transaction = client.transaction().await?;
18//!
19//! // Add commands to the transaction
20//! transaction.set("key1", "value1");
21//! transaction.set("key2", "value2");
22//! transaction.incr("counter");
23//!
24//! // Execute the transaction
25//! let results = transaction.exec().await?;
26//! println!("Transaction results: {:?}", results);
27//! # Ok(())
28//! # }
29//! ```
30
31use crate::commands::Command;
32use crate::core::{
33    error::{RedisError, RedisResult},
34    value::RespValue,
35};
36use std::collections::VecDeque;
37use std::sync::Arc;
38use tokio::sync::Mutex;
39
40/// A Redis transaction that executes commands atomically
41pub struct Transaction {
42    commands: VecDeque<Box<dyn TransactionCommand>>,
43    connection: Arc<Mutex<dyn TransactionExecutor + Send + Sync>>,
44    watched_keys: Vec<String>,
45    is_started: bool,
46}
47
48/// Trait for commands that can be used in transactions
49pub trait TransactionCommand: Send + Sync {
50    /// Get the command name
51    fn name(&self) -> &str;
52
53    /// Get the command arguments
54    fn args(&self) -> Vec<RespValue>;
55
56    /// Get the key(s) involved in this command (for WATCH)
57    fn key(&self) -> Option<String>;
58}
59
60/// Trait for executing transactions
61#[async_trait::async_trait]
62pub trait TransactionExecutor {
63    /// Start a transaction with MULTI
64    async fn multi(&mut self) -> RedisResult<()>;
65
66    /// Execute a command in the transaction
67    async fn queue_command(&mut self, command: Box<dyn TransactionCommand>) -> RedisResult<()>;
68
69    /// Execute the transaction with EXEC
70    async fn exec(&mut self) -> RedisResult<Vec<RespValue>>;
71
72    /// Discard the transaction with DISCARD
73    async fn discard(&mut self) -> RedisResult<()>;
74
75    /// Watch keys for changes
76    async fn watch(&mut self, keys: Vec<String>) -> RedisResult<()>;
77
78    /// Unwatch all keys
79    async fn unwatch(&mut self) -> RedisResult<()>;
80}
81
82impl Transaction {
83    /// Create a new transaction
84    pub fn new(connection: Arc<Mutex<dyn TransactionExecutor + Send + Sync>>) -> Self {
85        Self {
86            commands: VecDeque::new(),
87            connection,
88            watched_keys: Vec::new(),
89            is_started: false,
90        }
91    }
92
93    /// Watch keys for changes before starting the transaction
94    ///
95    /// If any watched key is modified before EXEC, the transaction will be discarded.
96    ///
97    /// # Examples
98    ///
99    /// ```no_run
100    /// # use redis_oxide::{Client, ConnectionConfig};
101    /// # #[tokio::main]
102    /// # async fn main() -> Result<(), Box<dyn std::error::Error>> {
103    /// # let config = ConnectionConfig::new("redis://localhost:6379");
104    /// # let client = Client::connect(config).await?;
105    /// let mut transaction = client.transaction().await?;
106    ///
107    /// // Watch keys before starting transaction
108    /// transaction.watch(vec!["balance".to_string(), "account".to_string()]).await?;
109    ///
110    /// // Add commands
111    /// transaction.set("balance", "100");
112    /// transaction.incr("account");
113    ///
114    /// // Execute - will fail if watched keys were modified
115    /// let results = transaction.exec().await?;
116    /// # Ok(())
117    /// # }
118    /// ```
119    pub async fn watch(&mut self, keys: Vec<String>) -> RedisResult<()> {
120        if self.is_started {
121            return Err(RedisError::Protocol("Cannot WATCH after MULTI".to_string()));
122        }
123
124        let mut connection = self.connection.lock().await;
125        connection.watch(keys.clone()).await?;
126        self.watched_keys.extend(keys);
127        Ok(())
128    }
129
130    /// Unwatch all previously watched keys
131    pub async fn unwatch(&mut self) -> RedisResult<()> {
132        let mut connection = self.connection.lock().await;
133        connection.unwatch().await?;
134        self.watched_keys.clear();
135        Ok(())
136    }
137
138    /// Add a command to the transaction
139    pub fn add_command(&mut self, command: Box<dyn TransactionCommand>) -> &mut Self {
140        self.commands.push_back(command);
141        self
142    }
143
144    /// Add a SET command to the transaction
145    pub fn set(&mut self, key: impl Into<String>, value: impl Into<String>) -> &mut Self {
146        use crate::commands::SetCommand;
147        let cmd = SetCommand::new(key.into(), value.into());
148        self.add_command(Box::new(cmd))
149    }
150
151    /// Add a GET command to the transaction
152    pub fn get(&mut self, key: impl Into<String>) -> &mut Self {
153        use crate::commands::GetCommand;
154        let cmd = GetCommand::new(key.into());
155        self.add_command(Box::new(cmd))
156    }
157
158    /// Add a DEL command to the transaction
159    pub fn del(&mut self, keys: Vec<String>) -> &mut Self {
160        use crate::commands::DelCommand;
161        let cmd = DelCommand::new(keys);
162        self.add_command(Box::new(cmd))
163    }
164
165    /// Add an INCR command to the transaction
166    pub fn incr(&mut self, key: impl Into<String>) -> &mut Self {
167        use crate::commands::IncrCommand;
168        let cmd = IncrCommand::new(key.into());
169        self.add_command(Box::new(cmd))
170    }
171
172    /// Add a DECR command to the transaction
173    pub fn decr(&mut self, key: impl Into<String>) -> &mut Self {
174        use crate::commands::DecrCommand;
175        let cmd = DecrCommand::new(key.into());
176        self.add_command(Box::new(cmd))
177    }
178
179    /// Add an INCRBY command to the transaction
180    pub fn incr_by(&mut self, key: impl Into<String>, increment: i64) -> &mut Self {
181        use crate::commands::IncrByCommand;
182        let cmd = IncrByCommand::new(key.into(), increment);
183        self.add_command(Box::new(cmd))
184    }
185
186    /// Add a DECRBY command to the transaction
187    pub fn decr_by(&mut self, key: impl Into<String>, decrement: i64) -> &mut Self {
188        use crate::commands::DecrByCommand;
189        let cmd = DecrByCommand::new(key.into(), decrement);
190        self.add_command(Box::new(cmd))
191    }
192
193    /// Add an EXISTS command to the transaction
194    pub fn exists(&mut self, keys: Vec<String>) -> &mut Self {
195        use crate::commands::ExistsCommand;
196        let cmd = ExistsCommand::new(keys);
197        self.add_command(Box::new(cmd))
198    }
199
200    /// Add an EXPIRE command to the transaction
201    pub fn expire(&mut self, key: impl Into<String>, seconds: std::time::Duration) -> &mut Self {
202        use crate::commands::ExpireCommand;
203        let cmd = ExpireCommand::new(key.into(), seconds);
204        self.add_command(Box::new(cmd))
205    }
206
207    /// Add a TTL command to the transaction
208    pub fn ttl(&mut self, key: impl Into<String>) -> &mut Self {
209        use crate::commands::TtlCommand;
210        let cmd = TtlCommand::new(key.into());
211        self.add_command(Box::new(cmd))
212    }
213
214    // Hash commands
215
216    /// Add an HGET command to the transaction
217    pub fn hget(&mut self, key: impl Into<String>, field: impl Into<String>) -> &mut Self {
218        use crate::commands::HGetCommand;
219        let cmd = HGetCommand::new(key.into(), field.into());
220        self.add_command(Box::new(cmd))
221    }
222
223    /// Add an HSET command to the transaction
224    pub fn hset(
225        &mut self,
226        key: impl Into<String>,
227        field: impl Into<String>,
228        value: impl Into<String>,
229    ) -> &mut Self {
230        use crate::commands::HSetCommand;
231        let cmd = HSetCommand::new(key.into(), field.into(), value.into());
232        self.add_command(Box::new(cmd))
233    }
234
235    /// Get the number of commands in the transaction
236    #[must_use]
237    pub fn len(&self) -> usize {
238        self.commands.len()
239    }
240
241    /// Check if the transaction is empty
242    #[must_use]
243    pub fn is_empty(&self) -> bool {
244        self.commands.is_empty()
245    }
246
247    /// Clear all commands from the transaction
248    pub fn clear(&mut self) {
249        self.commands.clear();
250    }
251
252    /// Execute the transaction
253    ///
254    /// This sends MULTI, queues all commands, and then executes them with EXEC.
255    /// Returns the results of all commands in the same order they were added.
256    ///
257    /// # Errors
258    ///
259    /// Returns an error if:
260    /// - The transaction is empty
261    /// - Network communication fails
262    /// - Any watched key was modified (returns empty result)
263    /// - Any command in the transaction fails
264    ///
265    /// # Examples
266    ///
267    /// ```no_run
268    /// # use redis_oxide::{Client, ConnectionConfig};
269    /// # #[tokio::main]
270    /// # async fn main() -> Result<(), Box<dyn std::error::Error>> {
271    /// # let config = ConnectionConfig::new("redis://localhost:6379");
272    /// # let client = Client::connect(config).await?;
273    /// let mut transaction = client.transaction().await?;
274    /// transaction.set("key1", "value1");
275    /// transaction.get("key1");
276    ///
277    /// let results = transaction.exec().await?;
278    /// assert_eq!(results.len(), 2);
279    /// # Ok(())
280    /// # }
281    /// ```
282    pub async fn exec(&mut self) -> RedisResult<Vec<RespValue>> {
283        if self.commands.is_empty() {
284            return Err(RedisError::Protocol("Transaction is empty".to_string()));
285        }
286
287        let mut connection = self.connection.lock().await;
288
289        // Start transaction if not already started
290        if !self.is_started {
291            connection.multi().await?;
292            self.is_started = true;
293        }
294
295        // Queue all commands
296        let commands: Vec<Box<dyn TransactionCommand>> = self.commands.drain(..).collect();
297        for command in commands {
298            connection.queue_command(command).await?;
299        }
300
301        // Execute the transaction
302        let results = connection.exec().await?;
303        self.is_started = false;
304
305        Ok(results)
306    }
307
308    /// Discard the transaction
309    ///
310    /// This cancels the transaction and discards all queued commands.
311    ///
312    /// # Examples
313    ///
314    /// ```no_run
315    /// # use redis_oxide::{Client, ConnectionConfig};
316    /// # #[tokio::main]
317    /// # async fn main() -> Result<(), Box<dyn std::error::Error>> {
318    /// # let config = ConnectionConfig::new("redis://localhost:6379");
319    /// # let client = Client::connect(config).await?;
320    /// let mut transaction = client.transaction().await?;
321    /// transaction.set("key1", "value1");
322    /// transaction.set("key2", "value2");
323    ///
324    /// // Cancel the transaction
325    /// transaction.discard().await?;
326    /// # Ok(())
327    /// # }
328    /// ```
329    pub async fn discard(&mut self) -> RedisResult<()> {
330        let mut connection = self.connection.lock().await;
331        if self.is_started {
332            connection.discard().await?;
333        } else if !self.watched_keys.is_empty() {
334            connection.unwatch().await?;
335        }
336        self.commands.clear();
337        self.watched_keys.clear();
338        self.is_started = false;
339        Ok(())
340    }
341}
342
343/// Transaction result wrapper for easier handling
344#[derive(Debug, Clone)]
345pub struct TransactionResult {
346    results: Vec<RespValue>,
347    index: usize,
348}
349
350impl TransactionResult {
351    /// Create a new transaction result
352    #[must_use]
353    pub fn new(results: Vec<RespValue>) -> Self {
354        Self { results, index: 0 }
355    }
356
357    /// Get the next result from the transaction
358    ///
359    /// # Errors
360    ///
361    /// Returns an error if there are no more results or type conversion fails.
362    pub fn next<T>(&mut self) -> RedisResult<T>
363    where
364        T: TryFrom<RespValue>,
365        T::Error: Into<RedisError>,
366    {
367        if self.index >= self.results.len() {
368            return Err(RedisError::Protocol(
369                "No more results in transaction".to_string(),
370            ));
371        }
372
373        let result = self.results[self.index].clone();
374        self.index += 1;
375
376        T::try_from(result).map_err(Into::into)
377    }
378
379    /// Get a result at a specific index
380    ///
381    /// # Errors
382    ///
383    /// Returns an error if the index is out of bounds or type conversion fails.
384    pub fn get<T>(&self, index: usize) -> RedisResult<T>
385    where
386        T: TryFrom<RespValue>,
387        T::Error: Into<RedisError>,
388    {
389        if index >= self.results.len() {
390            return Err(RedisError::Protocol(format!(
391                "Index {} out of bounds",
392                index
393            )));
394        }
395
396        let result = self.results[index].clone();
397        T::try_from(result).map_err(Into::into)
398    }
399
400    /// Get the number of results
401    #[must_use]
402    pub fn len(&self) -> usize {
403        self.results.len()
404    }
405
406    /// Check if there are no results
407    #[must_use]
408    pub fn is_empty(&self) -> bool {
409        self.results.is_empty()
410    }
411
412    /// Get all results as a vector
413    #[must_use]
414    pub fn into_results(self) -> Vec<RespValue> {
415        self.results
416    }
417}
418
419// Implement TransactionCommand for all command types
420impl TransactionCommand for crate::commands::GetCommand {
421    fn name(&self) -> &str {
422        self.command_name()
423    }
424
425    fn args(&self) -> Vec<RespValue> {
426        <Self as Command>::args(self)
427    }
428
429    fn key(&self) -> Option<String> {
430        Some(self.keys()[0].iter().map(|&b| b as char).collect())
431    }
432}
433
434impl TransactionCommand for crate::commands::SetCommand {
435    fn name(&self) -> &str {
436        self.command_name()
437    }
438
439    fn args(&self) -> Vec<RespValue> {
440        <Self as Command>::args(self)
441    }
442
443    fn key(&self) -> Option<String> {
444        Some(self.keys()[0].iter().map(|&b| b as char).collect())
445    }
446}
447
448impl TransactionCommand for crate::commands::DelCommand {
449    fn name(&self) -> &str {
450        self.command_name()
451    }
452
453    fn args(&self) -> Vec<RespValue> {
454        <Self as Command>::args(self)
455    }
456
457    fn key(&self) -> Option<String> {
458        if let Some(first_key) = self.keys().first() {
459            Some(first_key.iter().map(|&b| b as char).collect())
460        } else {
461            None
462        }
463    }
464}
465
466impl TransactionCommand for crate::commands::IncrCommand {
467    fn name(&self) -> &str {
468        self.command_name()
469    }
470
471    fn args(&self) -> Vec<RespValue> {
472        <Self as Command>::args(self)
473    }
474
475    fn key(&self) -> Option<String> {
476        Some(self.keys()[0].iter().map(|&b| b as char).collect())
477    }
478}
479
480impl TransactionCommand for crate::commands::DecrCommand {
481    fn name(&self) -> &str {
482        self.command_name()
483    }
484
485    fn args(&self) -> Vec<RespValue> {
486        <Self as Command>::args(self)
487    }
488
489    fn key(&self) -> Option<String> {
490        Some(self.keys()[0].iter().map(|&b| b as char).collect())
491    }
492}
493
494impl TransactionCommand for crate::commands::IncrByCommand {
495    fn name(&self) -> &str {
496        self.command_name()
497    }
498
499    fn args(&self) -> Vec<RespValue> {
500        <Self as Command>::args(self)
501    }
502
503    fn key(&self) -> Option<String> {
504        Some(self.keys()[0].iter().map(|&b| b as char).collect())
505    }
506}
507
508impl TransactionCommand for crate::commands::DecrByCommand {
509    fn name(&self) -> &str {
510        self.command_name()
511    }
512
513    fn args(&self) -> Vec<RespValue> {
514        <Self as Command>::args(self)
515    }
516
517    fn key(&self) -> Option<String> {
518        Some(self.keys()[0].iter().map(|&b| b as char).collect())
519    }
520}
521
522impl TransactionCommand for crate::commands::ExistsCommand {
523    fn name(&self) -> &str {
524        self.command_name()
525    }
526
527    fn args(&self) -> Vec<RespValue> {
528        <Self as Command>::args(self)
529    }
530
531    fn key(&self) -> Option<String> {
532        if let Some(first_key) = self.keys().first() {
533            Some(first_key.iter().map(|&b| b as char).collect())
534        } else {
535            None
536        }
537    }
538}
539
540impl TransactionCommand for crate::commands::ExpireCommand {
541    fn name(&self) -> &str {
542        self.command_name()
543    }
544
545    fn args(&self) -> Vec<RespValue> {
546        <Self as Command>::args(self)
547    }
548
549    fn key(&self) -> Option<String> {
550        Some(self.keys()[0].iter().map(|&b| b as char).collect())
551    }
552}
553
554impl TransactionCommand for crate::commands::TtlCommand {
555    fn name(&self) -> &str {
556        self.command_name()
557    }
558
559    fn args(&self) -> Vec<RespValue> {
560        <Self as Command>::args(self)
561    }
562
563    fn key(&self) -> Option<String> {
564        Some(self.keys()[0].iter().map(|&b| b as char).collect())
565    }
566}
567
568impl TransactionCommand for crate::commands::HGetCommand {
569    fn name(&self) -> &str {
570        self.command_name()
571    }
572
573    fn args(&self) -> Vec<RespValue> {
574        <Self as Command>::args(self)
575    }
576
577    fn key(&self) -> Option<String> {
578        Some(self.keys()[0].iter().map(|&b| b as char).collect())
579    }
580}
581
582impl TransactionCommand for crate::commands::HSetCommand {
583    fn name(&self) -> &str {
584        self.command_name()
585    }
586
587    fn args(&self) -> Vec<RespValue> {
588        <Self as Command>::args(self)
589    }
590
591    fn key(&self) -> Option<String> {
592        Some(self.keys()[0].iter().map(|&b| b as char).collect())
593    }
594}
595
596#[cfg(test)]
597mod tests {
598    use super::*;
599    use std::sync::Arc;
600    use tokio::sync::Mutex;
601
602    struct MockTransactionExecutor {
603        commands: Vec<String>,
604        multi_called: bool,
605        exec_called: bool,
606    }
607
608    impl MockTransactionExecutor {
609        fn new() -> Self {
610            Self {
611                commands: Vec::new(),
612                multi_called: false,
613                exec_called: false,
614            }
615        }
616    }
617
618    #[async_trait::async_trait]
619    impl TransactionExecutor for MockTransactionExecutor {
620        async fn multi(&mut self) -> RedisResult<()> {
621            self.multi_called = true;
622            Ok(())
623        }
624
625        async fn queue_command(&mut self, command: Box<dyn TransactionCommand>) -> RedisResult<()> {
626            self.commands.push(command.name().to_string());
627            Ok(())
628        }
629
630        async fn exec(&mut self) -> RedisResult<Vec<RespValue>> {
631            self.exec_called = true;
632            let mut results = Vec::new();
633            for _ in 0..self.commands.len() {
634                results.push(RespValue::SimpleString("OK".to_string()));
635            }
636            Ok(results)
637        }
638
639        async fn discard(&mut self) -> RedisResult<()> {
640            self.commands.clear();
641            self.multi_called = false;
642            Ok(())
643        }
644
645        async fn watch(&mut self, _keys: Vec<String>) -> RedisResult<()> {
646            Ok(())
647        }
648
649        async fn unwatch(&mut self) -> RedisResult<()> {
650            Ok(())
651        }
652    }
653
654    #[tokio::test]
655    async fn test_transaction_creation() {
656        let executor = MockTransactionExecutor::new();
657        let transaction = Transaction::new(Arc::new(Mutex::new(executor)));
658
659        assert!(transaction.is_empty());
660        assert_eq!(transaction.len(), 0);
661    }
662
663    #[tokio::test]
664    async fn test_transaction_add_commands() {
665        let executor = MockTransactionExecutor::new();
666        let mut transaction = Transaction::new(Arc::new(Mutex::new(executor)));
667
668        transaction.set("key1", "value1");
669        transaction.get("key1");
670
671        assert_eq!(transaction.len(), 2);
672        assert!(!transaction.is_empty());
673    }
674
675    #[tokio::test]
676    async fn test_transaction_exec() {
677        let executor = MockTransactionExecutor::new();
678        let mut transaction = Transaction::new(Arc::new(Mutex::new(executor)));
679
680        transaction.set("key1", "value1");
681        transaction.get("key1");
682
683        let results = transaction.exec().await.unwrap();
684        assert_eq!(results.len(), 2);
685        assert!(transaction.is_empty()); // Commands should be consumed
686    }
687
688    #[tokio::test]
689    async fn test_transaction_discard() {
690        let executor = MockTransactionExecutor::new();
691        let mut transaction = Transaction::new(Arc::new(Mutex::new(executor)));
692
693        transaction.set("key1", "value1");
694        transaction.get("key1");
695        assert_eq!(transaction.len(), 2);
696
697        transaction.discard().await.unwrap();
698        assert!(transaction.is_empty());
699    }
700
701    #[tokio::test]
702    async fn test_transaction_result() {
703        let results = vec![
704            RespValue::SimpleString("OK".to_string()),
705            RespValue::BulkString(b"value1".to_vec().into()),
706            RespValue::Integer(42),
707        ];
708
709        let mut transaction_result = TransactionResult::new(results);
710
711        assert_eq!(transaction_result.len(), 3);
712        assert!(!transaction_result.is_empty());
713
714        let first: String = transaction_result.next().unwrap();
715        assert_eq!(first, "OK");
716
717        let second: String = transaction_result.get(1).unwrap();
718        assert_eq!(second, "value1");
719    }
720}
redis_oxide/transaction.rs

redis_oxide/
transaction.rs
