oxana 2.1.3

A simple & fast job queue 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
80
81
82
83
84
85

use serde::{Deserialize, Serialize};
use std::time::Duration;
use tracing_subscriber::{EnvFilter, fmt, prelude::*};

#[derive(oxana::Registry)]
struct ComponentRegistry(oxana::ComponentRegistry<WorkerContext>);

#[derive(Debug, thiserror::Error)]
enum WorkerError {}

#[derive(Debug, Clone)]
struct WorkerContext {}

#[derive(Debug, Serialize, Deserialize, oxana::Job)]
struct OneSecJob {
    id: usize,
}

#[derive(oxana::Worker)]
struct OneSecWorker;

impl OneSecWorker {
    async fn process(&self, job: OneSecJob, _ctx: &oxana::JobContext) -> Result<(), WorkerError> {
        println!("job {} started", job.id);
        tokio::time::sleep(Duration::from_secs(1)).await;
        println!("job {} finished", job.id);
        Ok(())
    }
}

#[derive(Serialize, oxana::Queue)]
#[oxana(key = "dynamic_concurrency", concurrency = Dynamic(1))]
struct DynamicConcurrencyQueue;

#[tokio::main]
pub async fn main() -> Result<(), oxana::OxanaError> {
    tracing_subscriber::registry()
        .with(fmt::layer())
        .with(EnvFilter::from_default_env())
        .init();

    let ctx = WorkerContext {};
    let storage = oxana::Storage::builder().build_from_env()?;
    let runtime = storage
        .runtime(ctx)
        .register::<ComponentRegistry>()
        .exit_when_processed(30);

    storage.reset_queue_config(DynamicConcurrencyQueue).await?;

    for id in 1..=30 {
        storage
            .enqueue(DynamicConcurrencyQueue, OneSecJob { id })
            .await?;
    }

    let update_storage = storage.clone();
    let handle = tokio::runtime::Handle::current();
    let update_thread = std::thread::spawn(move || {
        std::thread::sleep(Duration::from_secs(5));

        handle.block_on(async move {
            update_storage
                .set_queue_concurrency(DynamicConcurrencyQueue, 0)
                .await?;
            println!("processing has been paused");

            tokio::time::sleep(Duration::from_secs(10)).await;
            println!("processing is resuming");

            update_storage
                .set_queue_concurrency(DynamicConcurrencyQueue, 10)
                .await
        })
    });

    let run_result = runtime.run().await;
    let update_result = update_thread.join().map_err(|_| {
        oxana::OxanaError::GenericError("concurrency update thread panicked".into())
    })?;
    update_result?;
    run_result?;

    Ok(())
}