dope_fiber/net/client/
connect.rs

1use std::{ops::ControlFlow, task};
2
3use dope::{
4    core::driver::route::{Identity, KeyTag},
5    net::{self, wire},
6};
7
8use crate::{
9    abi::{self, machine},
10    context,
11    manifold::client::attempt::queue::mailbox,
12    net::{Io, client},
13};
14
15enum Stage<'scope, 'w, 'd: 'w, T: net::Transport, W: wire::Spec, ID: Identity> {
16    Init(T::Addr),
17    Ticket(client::Ticket<'scope, 'w, 'd, T, W, ID>),
18}
19
20/// A connect operation that owns any waker registration made for its dial ticket.
21#[must_use = "a fiber does nothing unless it is driven"]
22pub struct Connect<'scope, 'w, 'd: 'w, T: net::Transport, W: wire::Spec, ID: Identity = KeyTag<0>> {
23    host: client::Handle<'scope, 'w, 'd, T, W, ID>,
24    stage: machine::Machine<Stage<'scope, 'w, 'd, T, W, ID>>,
25}
26
27impl<'scope, 'w, 'd: 'w, T: net::Transport, W: wire::Spec, ID: Identity>
28    Connect<'scope, 'w, 'd, T, W, ID>
29{
30    pub(super) fn new(host: client::Handle<'scope, 'w, 'd, T, W, ID>, addr: T::Addr) -> Self {
31        Self {
32            host,
33            stage: machine::Machine::new(Stage::Init(addr)),
34        }
35    }
36}
37
38impl<T: net::Transport, W: wire::Spec, ID: Identity> Unpin for Connect<'_, '_, '_, T, W, ID> {}
39
40impl<T: net::Transport, W: wire::Spec, ID: Identity> Drop for Connect<'_, '_, '_, T, W, ID> {
41    fn drop(&mut self) {
42        let Some(Stage::Ticket(lease)) = self.stage.take() else {
43            return;
44        };
45        if let mailbox::Cancellation::Settled(outcome) = lease.cancel() {
46            self.host.port.reclaim(outcome);
47        }
48    }
49}
50
51impl<'scope, 'w, 'd: 'w, T: net::Transport, W: wire::Spec, ID: Identity> abi::Fiber<'d>
52    for Connect<'scope, 'w, 'd, T, W, ID>
53{
54    type Output = client::Connected<'scope, 'w, 'd, W, ID>;
55
56    fn poll(call: context::PollCall<'_, '_, 'd, Self>) -> task::Poll<Self::Output> {
57        let (this, cx) = call.into_parts();
58        let this = this.get_mut();
59        let wake = cx.as_ref().completion();
60        this.stage.step(|stage| {
61            let lease = match stage {
62                Stage::Ticket(lease) => lease,
63                Stage::Init(addr) => match this.host.port.dial(addr) {
64                    Ok(lease) => lease,
65                    Err(error) => return ControlFlow::Break(Err(error)),
66                },
67            };
68            let (outcome, lease) = match lease.poll(wake) {
69                mailbox::Poll::Ready(outcome, lease) => (outcome, lease),
70                mailbox::Poll::Pending(lease) => {
71                    return ControlFlow::Continue(Stage::Ticket(lease));
72                }
73            };
74            let error = match outcome {
75                client::Outcome::Connected(conn) => {
76                    let io = Io::new(&this.host.port.conns, conn);
77                    lease.commit();
78                    return ControlFlow::Break(Ok(io));
79                }
80                client::Outcome::Activation => client::ConnectError::Activation,
81                client::Outcome::Transport(cause) => client::ConnectError::Transport(cause),
82                client::Outcome::Open(cause) => client::ConnectError::Open(cause),
83            };
84            lease.cancel();
85            ControlFlow::Break(Err(error))
86        })
87    }
88}
dope_fiber/net/client/connect.rs

dope_fiber/net/client/
connect.rs
