floxide-redis 3.2.2

Redis implementation for Floxide distributed workflow system
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79

mod helpers;

use async_trait::async_trait;
use floxide_core::distributed::WorkQueue;
use floxide_redis::{RedisClient, RedisConfig, RedisWorkQueue};

#[derive(Debug, Clone, Default, serde::Serialize, serde::Deserialize)]
struct DummyContext;

#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
struct DummyWorkItem {
    pub id: String,
}

impl std::fmt::Display for DummyWorkItem {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        write!(f, "{}", self.id)
    }
}

#[async_trait]
impl floxide_core::workflow::WorkItem for DummyWorkItem {
    fn instance_id(&self) -> String {
        self.id.clone()
    }
    fn is_terminal(&self) -> bool {
        false
    }
}

#[tokio::test]
async fn test_work_queue_roundtrip() {
    helpers::init_tracing();
    let mut redis = helpers::TestRedis::start().await;
    let config = RedisConfig::new(redis.redis_url());
    let client = RedisClient::new(config).await.expect("redis client");
    let queue = RedisWorkQueue::<DummyWorkItem>::new(client);

    let run_id = "run1";
    let item1 = DummyWorkItem {
        id: "item1".to_string(),
    };
    let item2 = DummyWorkItem {
        id: "item2".to_string(),
    };

    // Enqueue
    WorkQueue::<DummyContext, DummyWorkItem>::enqueue(&queue, run_id, item1.clone())
        .await
        .expect("enqueue");
    WorkQueue::<DummyContext, DummyWorkItem>::enqueue(&queue, run_id, item2.clone())
        .await
        .expect("enqueue");

    // Pending work
    let pending = WorkQueue::<DummyContext, DummyWorkItem>::pending_work(&queue, run_id)
        .await
        .expect("pending");
    assert_eq!(pending.len(), 2);

    // Dequeue
    let (deq_run_id, deq_item) = WorkQueue::<DummyContext, DummyWorkItem>::dequeue(&queue)
        .await
        .expect("dequeue")
        .expect("item");
    assert_eq!(deq_run_id, run_id);
    assert!(deq_item == item1 || deq_item == item2);

    // Purge
    WorkQueue::<DummyContext, DummyWorkItem>::purge_run(&queue, run_id)
        .await
        .expect("purge");
    let pending = WorkQueue::<DummyContext, DummyWorkItem>::pending_work(&queue, run_id)
        .await
        .expect("pending");
    assert!(pending.is_empty());
    tracing::info!(?pending, "Work queue roundtrip successful");
    redis.cleanup().await;
}