cartel-pg 0.27.0

Async PostgreSQL driver for the dope runtime
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
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103

use crate::reply::Error;
use crate::{api, schema};
use api::{Response, request, task};
use cartel_core::deadline;
use dope::{
    core::driver::{route::KeyTag, schedule::clock::timer},
    manifold::client::conn,
};
use dope_fiber::{abi, context, wait};
use std::{pin, task::Poll};
use task::lease::{Lease, Recover};

#[pin_project::pin_project]
pub(in crate::api) struct Stateful<'app, 'g, 'r, 's, 'b, 'c, 'd: 'app, B: ?Sized, I, const ID: u8>
where
    I: schema::QuerySet + 'd,
{
    command: task::Core<
        'app,
        'g,
        'r,
        's,
        'b,
        'c,
        'd,
        B,
        I,
        Response<'app, 'g, 'r, 's, 'b, 'c, 'd, api::ExtractUnit<ID>>,
        ID,
        (),
        conn::Id<'d, KeyTag<ID>>,
        Recover,
    >,
    #[pin]
    deadline: Result<deadline::Guard<'d>, timer::Exhausted>,
    #[pin]
    waiter: wait::Waiter<'app, 'd>,
}

impl<'app, 'g, 'r, 's, 'b, 'c, 'd: 'app, B: ?Sized, I: schema::QuerySet + 'd, const ID: u8>
    Stateful<'app, 'g, 'r, 's, 'b, 'c, 'd, B, I, ID>
{
    pub(in crate::api) fn new(
        client: api::Client<'app, 'g, 'r, 's, 'b, 'c, 'd, B, I, ID>,
        target: Option<api::Target<'d, ID>>,
        req: request::Request<'_>,
    ) -> Self {
        let core = client
            .raw_core_as::<api::ExtractUnit<ID>, conn::Id<'d, KeyTag<ID>>, Recover>(target, req);
        Self {
            command: core,
            deadline: client.port.lifecycle.deadline.guard(),
            waiter: wait::Waiter::new(),
        }
    }
}

impl<'app, 'g, 'r, 's, 'b, 'c, 'd: 'app, B: ?Sized, I: schema::QuerySet + 'd, const ID: u8>
    abi::Fiber<'d> for Stateful<'app, 'g, 'r, 's, 'b, 'c, 'd, B, I, ID>
{
    type Output = Result<
        Lease<'app, 'g, 'r, 's, 'b, 'c, 'd, B, I, conn::Id<'d, KeyTag<ID>>, Recover, ID>,
        Error<'d, ID>,
    >;

    fn poll(call: context::PollCall<'_, '_, 'd, Self>) -> Poll<Self::Output> {
        let (this, mut ctx) = call.into_parts();
        let mut this = this.project();
        let mut guard = match this.deadline.as_mut().get_mut() {
            Ok(guard) => pin::Pin::new(guard),
            Err(error) => {
                this.waiter.as_ref().unregister();
                return Poll::Ready(Err(this.command.abort(crate::Error::Deadline(*error))));
            }
        };
        let phase = this.command.phase();
        if let Poll::Ready(time) = guard.as_mut().poll_expired(ctx.as_mut(), phase) {
            this.waiter.as_ref().unregister();
            return Poll::Ready(Err(this.command.abort(crate::Error::RequestTimeout(time))));
        }
        match this.command.poll(
            ctx.as_mut(),
            this.waiter.as_ref(),
            Lease::admit,
            task::Core::complete,
        ) {
            Poll::Pending => Poll::Pending,
            Poll::Ready(Err(error)) => {
                guard.as_mut().complete();
                Poll::Ready(Err(error))
            }
            Poll::Ready(Ok((Ok(()), lease))) => {
                guard.as_mut().complete();
                Poll::Ready(Ok(lease))
            }
            Poll::Ready(Ok((Err(error), lease))) => {
                lease.settle();
                guard.as_mut().complete();
                Poll::Ready(Err(error))
            }
        }
    }
}