dope_fiber/
file.rs

1//! Owned file operations backed by an installed manifold Reader.
2
3use std::task;
4
5use o3::mem::Rejected;
6
7use crate::{
8    abi::{self, machine},
9    context,
10    manifold::file::read,
11};
12
13enum Stage<'p, 'd> {
14    Request(read::Request<'d>),
15    Ticket(read::Ticket<'p, 'd>),
16}
17
18/// Cancellation drops only the reply ticket. The Reader keeps native inputs.
19#[must_use = "a fiber does nothing unless it is driven"]
20pub struct Operation<'p, 'd> {
21    port: &'p read::Port<'d>,
22    stage: machine::Machine<Stage<'p, 'd>>,
23}
24
25impl<'p, 'd> Operation<'p, 'd> {
26    pub fn new(port: &'p read::Port<'d>, request: read::Request<'d>) -> Self {
27        Self {
28            port,
29            stage: machine::Machine::new(Stage::Request(request)),
30        }
31    }
32}
33
34impl<'d> abi::Fiber<'d> for Operation<'_, 'd> {
35    type Output = Result<read::Completion<'d>, Rejected<read::Request<'d>, read::Admission>>;
36
37    fn poll(call: context::PollCall<'_, '_, 'd, Self>) -> task::Poll<Self::Output> {
38        let (this, cx) = call.into_parts();
39        let this = this.get_mut();
40        let waiter = cx.as_ref().completion();
41        this.stage.step(|stage| {
42            let ticket = match stage {
43                Stage::Ticket(ticket) => ticket,
44                Stage::Request(request) => match this.port.try_request(request) {
45                    Ok(ticket) => ticket,
46                    Err(cause) => return machine::Step::Complete(task::Poll::Ready(Err(cause))),
47                },
48            };
49            match ticket.poll(waiter) {
50                read::Poll::Pending(ticket) => {
51                    machine::Step::Suspend(Stage::Ticket(ticket), task::Poll::Pending)
52                }
53                read::Poll::Ready(completion) => {
54                    machine::Step::Complete(task::Poll::Ready(Ok(completion)))
55                }
56            }
57        })
58    }
59}
dope_fiber/file.rs

dope_fiber/
file.rs
