dge-gen 0.2.0

Crate for building a computational graph to be executed distributely
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

{% include "part_comment.rs" %}

{% include "part_common_import.rs" %}

use std::collections::VecDeque;
use std::sync::Arc;
use tokio::sync::RwLock;
use dge_runtime::component::poll::Jobs;
use dge_runtime::component::poll::new_job;
use dge_runtime::component::poll::poll_forever;

#[rustfmt::skip]
#[tokio::main(worker_threads = 10)]
pub(crate) async fn main() -> Result<()> {
    // load existing jobs
    let jobs = load_jobs().await?;

    // start a thread to poll the jobs
    tokio::spawn(poll_forever(
        {{ behaviour_module }}::get_capacity(),
        jobs.clone(),

        // these are used when do the actual checking
        {{ behaviour_module }}::check,
        {{ accept_failure }},
        {{ rmq_options.get_rmq_uri }},
        "{{ rmq_options.work_exchange }}",
        {{ output_queue }},
    ));

    let rmq_uri = {{ rmq_options.get_rmq_uri }}();
    let () = dge_runtime::rmq::consume_forever(
        &rmq_uri,
        {{ input_queue }},
        handler,
        jobs,
        {{ prefetch_count }},
    ).await;

    Ok(())
}

async fn load_jobs() -> Result<Jobs<{{ type_input }}>> {
    let jobs = Arc::new(RwLock::new(VecDeque::new()));

    let write_jobs = jobs.clone();
    let mut write_jobs = write_jobs.write().await;

    info!("loading messages");
    let msgs = {{ behaviour_module }}::init().await;

    info!("loaded {} messages, adding them to the job queue", msgs.len());
    for msg in msgs {
        let job = new_job(msg);
        write_jobs.push_back(job);
    }
    info!("jobs added to the job queue");

    Ok(jobs)
}


#[rustfmt::skip]
async fn handler(
    jobs: Jobs<{{ type_input }}>,
    _channel: Channel,
    msg: {{ type_input }},
) -> Result<Responsibility>
{
    dge_runtime::add_to_jobs!(
        jobs = jobs,
        msg = msg,
        save_msg = {{ behaviour_module }}::save_msg,
    )
}