dope_fiber/net/client/
connect.rs

1use std::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) => {
66                        return machine::Step::Complete(task::Poll::Ready(Err(error)));
67                    }
68                },
69            };
70            let (outcome, lease) = match lease.poll(wake) {
71                mailbox::Poll::Ready(outcome, lease) => (outcome, lease),
72                mailbox::Poll::Pending(lease) => {
73                    return machine::Step::Suspend(Stage::Ticket(lease), task::Poll::Pending);
74                }
75            };
76            let error = match outcome {
77                client::Outcome::Connected(conn) => {
78                    let io = Io::new(&this.host.port.conns, conn);
79                    lease.commit();
80                    return machine::Step::Complete(task::Poll::Ready(Ok(io)));
81                }
82                client::Outcome::Activation => client::ConnectError::Activation,
83                client::Outcome::Transport(cause) => client::ConnectError::Transport(cause),
84                client::Outcome::Open(cause) => client::ConnectError::Open(cause),
85            };
86            lease.cancel();
87            machine::Step::Complete(task::Poll::Ready(Err(error)))
88        })
89    }
90}
dope_fiber/net/client/connect.rs

dope_fiber/net/client/
connect.rs
