saddle-runtime 0.3.24

Saddle managed asynchronous runtime and lifecycle
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

//! Framework consumer shape; not business API or a second product process.
use saddle_observability::{EarlyRequestContext, FrameworkRequestFailure, RequestDiagnosticScope};
use saddle_runtime::request_task::{
    admitted_request_task, rejection_request_task, RequestTaskJoin,
};
use std::collections::HashMap;

fn early() -> RequestDiagnosticScope<'static> {
    RequestDiagnosticScope::early(None, EarlyRequestContext::socket_accepted("configured-app"))
}

// The actual Service already owns this dispatch; RP never mints it.
#[allow(dead_code)]
async fn admitted(owner: saddle_runtime::profusegw::ProfuseGwManagedDispatch) {
    let (future, ticket) = admitted_request_task(owner, early(), None, |owner, _context| {
        Box::pin(async move {
            let _ = owner.deadline_unix_ms();
            Ok::<_, FrameworkRequestFailure<std::io::Error>>(())
        })
    });
    let mut tasks = tokio::task::JoinSet::new();
    let abort = tasks.spawn(future);
    let ticket = ticket.bind_spawned(abort.id()).ok().unwrap();
    let (owner, outcome) = ticket
        .complete(tasks.join_next().await.unwrap())
        .ok()
        .unwrap();
    let terminal = outcome.finish();
    assert!(terminal.outcome().result().unwrap().is_ok());
    drop(terminal);
    owner.cancel();
}

async fn run() {
    // Reuse the existing bounded rejection policy. Tickets track the same set,
    // not an extra scheduler/queue; remove them when each join result is consumed.
    const MAX_REJECTION_WRITES: usize = 16;
    let mut tasks = tokio::task::JoinSet::new();
    let mut tickets: HashMap<tokio::task::Id, RequestTaskJoin> = HashMap::new();
    assert!(tasks.len() < MAX_REJECTION_WRITES);
    let (future, ticket) = rejection_request_task(early(), None, |context| async move {
        Err::<(), _>(context.scope().fail(
            std::io::Error::from(std::io::ErrorKind::UnexpectedEof),
            saddle_core::DiagnosticCategory::UnexpectedError,
            saddle_core::BoundedDiagnosticCause::new(
                saddle_core::DiagnosticStage::BackgroundTask,
                saddle_core::DiagnosticCode::new("example.request_task_read").unwrap(),
            ),
        ))
    });
    let abort = tasks.spawn(future);
    tickets.insert(abort.id(), ticket.bind_spawned(abort.id()).ok().unwrap());
    while let Some(result) = tasks.join_next_with_id().await {
        let (id, result) = match result {
            Ok((id, result)) => (id, Ok(result)),
            Err(error) => (error.id(), Err(error)),
        };
        let terminal = tickets
            .remove(&id)
            .unwrap()
            .complete(result)
            .ok()
            .unwrap()
            .finish();
        let failure = terminal.outcome().result().unwrap().as_ref().err().unwrap();
        assert_eq!(failure.error().kind(), std::io::ErrorKind::UnexpectedEof);
        assert_eq!(
            failure.submission(),
            saddle_observability::DiagnosticSubmission::OutputUnavailable
        );
    }
    assert!(tickets.is_empty());
}

fn main() {
    tokio::runtime::Builder::new_current_thread()
        .enable_all()
        .build()
        .unwrap()
        .block_on(run());
}