libp2p_pubsub_common/service/
context.rs

1use std::collections::VecDeque;
2use std::task::{Context as TaskContext, Poll};
3
4use crate::service::context_handles::PollCtx;
5use crate::service::service_trait::Service;
6
7/// A service context is a wrapper around a service that can be used to send events to it and poll
8/// it for events.
9///
10/// See [`BufferedContext`] documentation for an implementation that provides a mailbox for input
11/// events and a mailbox for output events.
12pub trait ServiceContext {
13    /// The input event type.
14    type InEvent: 'static;
15    /// The output event type.
16    type OutEvent: 'static;
17
18    /// Sends a message to the service context unconditionally, ignoring any potential errors.
19    fn do_send(&mut self, ev: Self::InEvent);
20
21    /// Polls the service context for events.
22    fn poll(&mut self, cx: &mut TaskContext<'_>) -> Poll<Self::OutEvent>;
23}
24
25/// A buffered service context.
26///
27/// This [`ServiceContext`] implementation is a wrapper around a service that provides a mailbox
28/// for input events and a mailbox for output events. The mailboxes are implemented as FIFO queues
29/// based on [`VecDeque`].
30///
31/// The input mailbox acts as a buffer for the events that are sent to the service. The events are
32/// enqueued into the mailbox by calling the [`do_send`](#method.do_send) method. The mailbox
33/// events are processed by the service on the next [`BufferedContext::poll`] call (see
34/// [`poll`](#method.poll) for more details).
35///
36/// The service context implements `Deref` trait to the inner service, so it can be used as the
37/// wrapped service itself.
38pub struct BufferedContext<S: Service> {
39    service: S,
40    inbox: VecDeque<S::InEvent>,
41    outbox: VecDeque<S::OutEvent>,
42}
43
44/// Public API,
45impl<S: Service> BufferedContext<S> {
46    /// Create a new service context for the given service.
47    pub fn new(service: S) -> Self {
48        Self {
49            service,
50            inbox: VecDeque::new(),
51            outbox: VecDeque::new(),
52        }
53    }
54
55    /// Get a reference to the inner service.
56    pub fn service(&self) -> &S {
57        &self.service
58    }
59}
60
61impl<S: Service> ServiceContext for BufferedContext<S> {
62    type InEvent = S::InEvent;
63    type OutEvent = S::OutEvent;
64
65    /// Sends a message unconditionally, ignoring any potential errors.
66    ///
67    /// The event will be enqueued into the service input mailbox and will be processed by the
68    /// service on the next [`BufferedContext::poll`] call. See [`poll`](#method.poll) for more details.
69    fn do_send(&mut self, ev: S::InEvent) {
70        self.inbox.push_back(ev);
71    }
72
73    /// Poll the service for events.
74    ///
75    /// The service polling process consists of the following steps:
76    ///  1. Poll the service for events. This triggers the [`Service::poll`] method, allowing the
77    ///     service to emit events.
78    ///
79    ///     If the `poll` method returns a `Poll::Ready` event, the polling process will stop any
80    ///     further polling and the event will be returned to the downstream service.
81    ///  2. Process all the service output mailbox events. This drains the output mailbox and
82    ///     returns the events to the downstream service.
83    fn poll(&mut self, cx: &mut TaskContext<'_>) -> Poll<S::OutEvent> {
84        // Poll the service for events.
85        let svc_cx = PollCtx::new(&mut self.inbox, &mut self.outbox);
86        if let Poll::Ready(event) = self.service.poll(svc_cx, cx) {
87            return Poll::Ready(event);
88        }
89
90        // Process the outbox events.
91        if let Some(ev) = self.outbox.pop_front() {
92            return Poll::Ready(ev);
93        }
94
95        Poll::Pending
96    }
97}
98
99impl<S: Service + Default> Default for BufferedContext<S> {
100    fn default() -> Self {
101        Self::new(Default::default())
102    }
103}
104
105impl<S: Service> std::ops::Deref for BufferedContext<S> {
106    type Target = S;
107
108    fn deref(&self) -> &Self::Target {
109        &self.service
110    }
111}
libp2p_pubsub_common/service/context.rs

libp2p_pubsub_common/service/
context.rs
