ocpp_client/
runtime.rs

1//! Runtime abstraction so the engine (`Client<E>`) doesn't hard-depend on tokio. `Executor`
2//! spawns the background read loop and per-handler tasks; `Timer` drives request/ping
3//! timeouts. Both are dyn-safe (boxed-future style) so `Client<E>` stays generic over one
4//! type parameter only, the same way `TransportSink`/`TransportStream` are already boxed
5//! instead of threaded through as generics. `tokio-runtime` (see `runtime::tokio`) provides
6//! the default std impls; embedded users supply their own (e.g. backed by
7//! `embassy-executor`/`embassy-time`).
8
9use alloc::boxed::Box;
10use core::future::Future;
11use core::pin::Pin;
12use core::task::Poll;
13use core::time::Duration;
14
15#[cfg(feature = "tokio-runtime")]
16pub mod tokio;
17
18/// Spawns futures onto a background executor. Implementations must actually run the future
19/// to completion independently of the caller awaiting anything - `Client::from_transport`'s
20/// read loop, `on()`'s per-action handler loop, and `on_ping()`'s subscriber loop all rely on
21/// `spawn` to keep running in the background.
22pub trait Executor: Send + Sync + 'static {
23    fn spawn(&self, future: Pin<Box<dyn Future<Output = ()> + Send>>);
24}
25
26/// Produces timer delays. `with_timeout` (below) is built on top of this single dyn-safe
27/// method rather than a generic `timeout<F>` method, so `Timer` itself stays object-safe.
28pub trait Timer: Send + Sync + 'static {
29    fn delay<'a>(&'a self, duration: Duration) -> Pin<Box<dyn Future<Output = ()> + Send + 'a>>;
30}
31
32/// Returned when a timeout elapses before the future it was racing resolves.
33///
34/// Produced by this crate's internal `with_timeout` helper; it is public because it surfaces
35/// through `Client`'s API, not because callers construct it.
36#[derive(Debug, Clone, Copy, PartialEq, Eq)]
37pub struct Elapsed;
38
39/// Returned when a cancellation signal fires before the future it was racing resolves.
40///
41/// Produced by this crate's internal `with_cancel` helper - see [`Elapsed`] for why it is public.
42#[derive(Debug, Clone, Copy, PartialEq, Eq)]
43pub struct Cancelled;
44
45/// Race `fut` against `cancel`, resolving to whichever finishes first - the same hand-rolled
46/// `poll_fn` shape as [`with_timeout`], but cancelled by another future rather than a timer.
47///
48/// `fut` is polled first, so a future that is already ready wins even if `cancel` is too. When
49/// `cancel` wins, `fut` is dropped mid-poll; every caller is responsible for only passing
50/// futures that tolerate that. The read loop's use of this is why `TransportStream::recv` is
51/// documented as having to be cancel-safe.
52pub(crate) async fn with_cancel<F: Future, C: Future>(
53    fut: F,
54    cancel: C,
55) -> Result<F::Output, Cancelled> {
56    let mut fut = core::pin::pin!(fut);
57    let mut cancel = core::pin::pin!(cancel);
58    core::future::poll_fn(move |cx| {
59        if let Poll::Ready(value) = fut.as_mut().poll(cx) {
60            return Poll::Ready(Ok(value));
61        }
62        if cancel.as_mut().poll(cx).is_ready() {
63            return Poll::Ready(Err(Cancelled));
64        }
65        Poll::Pending
66    })
67    .await
68}
69
70/// Race `fut` against `timer.delay(duration)`, by hand - no `futures::select`/extra
71/// dependency needed, just polling both each wake via `core::future::poll_fn`.
72pub(crate) async fn with_timeout<F: Future>(
73    timer: &dyn Timer,
74    duration: Duration,
75    fut: F,
76) -> Result<F::Output, Elapsed> {
77    let mut fut = core::pin::pin!(fut);
78    let mut delay = timer.delay(duration);
79    core::future::poll_fn(move |cx| {
80        if let Poll::Ready(value) = fut.as_mut().poll(cx) {
81            return Poll::Ready(Ok(value));
82        }
83        if let Poll::Ready(()) = delay.as_mut().poll(cx) {
84            return Poll::Ready(Err(Elapsed));
85        }
86        Poll::Pending
87    })
88    .await
89}
ocpp_client/runtime.rs

ocpp_client/
runtime.rs
