1use std::{ops::ControlFlow, task};
4
5use o3::mem::{Rejected, Return};
6
7use crate::{
8 abi::{self, machine},
9 context,
10 manifold::file::read,
11};
12
13enum Stage<'p, 'd, R: Return> {
14 Request(read::Request<R>),
15 Ticket(read::Ticket<'p, 'd, R>),
16}
17
18#[must_use = "a fiber does nothing unless it is driven"]
20pub struct Operation<'p, 'd, R: Return = read::Release<'d>> {
21 port: &'p read::Port<'d, R>,
22 stage: machine::Machine<Stage<'p, 'd, R>>,
23}
24
25impl<'p, 'd, R: Return> Operation<'p, 'd, R> {
26 pub fn new(port: &'p read::Port<'d, R>, request: read::Request<R>) -> Self {
27 Self {
28 port,
29 stage: machine::Machine::new(Stage::Request(request)),
30 }
31 }
32}
33
34impl<R: Return> Unpin for Operation<'_, '_, R> {}
36
37impl<'d, R: Return<Item = read::Buffer<'d>>> abi::Fiber<'d> for Operation<'_, 'd, R> {
38 type Output = Result<read::Completion<R>, Rejected<read::Request<R>, read::Admission>>;
39
40 fn poll(call: context::PollCall<'_, '_, 'd, Self>) -> task::Poll<Self::Output> {
41 let (this, cx) = call.into_parts();
42 let this = this.get_mut();
43 let waiter = cx.as_ref().completion();
44 this.stage.step(|stage| {
45 let ticket = match stage {
46 Stage::Ticket(ticket) => ticket,
47 Stage::Request(request) => match this.port.try_request(request) {
48 Ok(ticket) => ticket,
49 Err(cause) => return ControlFlow::Break(Err(cause)),
50 },
51 };
52 match ticket.poll(waiter) {
53 read::Poll::Pending(ticket) => ControlFlow::Continue(Stage::Ticket(ticket)),
54 read::Poll::Ready(completion) => ControlFlow::Break(Ok(completion)),
55 }
56 })
57 }
58}