syncweb_core/sync/
intents.rs

1use std::{
2    pin::Pin,
3    task::{Context, Poll},
4};
5
6use n0_future::{Sink, Stream};
7use tokio::sync::mpsc;
8
9use super::TransferStats;
10
11/// Commands accepted by a running synchronization intent.
12#[derive(Clone, Copy, Debug, Eq, PartialEq)]
13#[non_exhaustive]
14pub enum SyncCommand {
15    Pause,
16    Resume,
17    Cancel,
18}
19
20/// Progress and lifecycle events emitted by an intent.
21#[derive(Clone, Debug, Eq, PartialEq)]
22#[non_exhaustive]
23pub enum SyncEvent {
24    Started,
25    Progress { completed: u64, total: Option<u64> },
26    Stats(TransferStats),
27    Paused,
28    Resumed,
29    Cancelled,
30    Finished,
31    Failed(String),
32}
33
34/// Control channel for a running synchronization intent.
35#[derive(Clone, Debug)]
36pub struct IntentControl {
37    commands: mpsc::UnboundedSender<SyncCommand>,
38}
39
40impl IntentControl {
41    /// Request that the intent pause network synchronization.
42    ///
43    /// # Errors
44    ///
45    /// Returns an error if the intent has already stopped.
46    pub fn pause(&self) -> Result<(), mpsc::error::SendError<SyncCommand>> {
47        self.commands.send(SyncCommand::Pause)
48    }
49
50    /// Request that the intent resume network synchronization.
51    ///
52    /// # Errors
53    ///
54    /// Returns an error if the intent has already stopped.
55    pub fn resume(&self) -> Result<(), mpsc::error::SendError<SyncCommand>> {
56        self.commands.send(SyncCommand::Resume)
57    }
58
59    /// Request that the intent stop.
60    ///
61    /// # Errors
62    ///
63    /// Returns an error if the intent has already stopped.
64    pub fn cancel(&self) -> Result<(), mpsc::error::SendError<SyncCommand>> {
65        self.commands.send(SyncCommand::Cancel)
66    }
67}
68
69/// Bidirectional handle for a synchronization operation.
70///
71/// It implements both halves of the futures API: callers can consume
72/// progress events as a stream and send control commands as a sink.
73pub struct IntentHandle {
74    events: mpsc::UnboundedReceiver<SyncEvent>,
75    commands: mpsc::UnboundedSender<SyncCommand>,
76}
77
78impl std::fmt::Debug for IntentHandle {
79    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
80        formatter.debug_struct("IntentHandle").finish_non_exhaustive()
81    }
82}
83
84impl IntentHandle {
85    #[must_use]
86    pub fn channel() -> (
87        mpsc::UnboundedSender<SyncEvent>,
88        mpsc::UnboundedReceiver<SyncCommand>,
89        Self,
90    ) {
91        let (event_sender, events) = mpsc::unbounded_channel();
92        let (commands, command_receiver) = mpsc::unbounded_channel();
93        (event_sender, command_receiver, Self { events, commands })
94    }
95
96    /// Request a pause without going through the Sink trait.
97    /// # Errors
98    ///
99    /// Returns an error if the receiving end of the channel has been dropped.
100    pub fn pause(&self) -> Result<(), mpsc::error::SendError<SyncCommand>> {
101        self.commands.send(SyncCommand::Pause)
102    }
103
104    /// Request resumption without going through the Sink trait.
105    /// # Errors
106    ///
107    /// Returns an error if the receiving end of the channel has been dropped.
108    pub fn resume(&self) -> Result<(), mpsc::error::SendError<SyncCommand>> {
109        self.commands.send(SyncCommand::Resume)
110    }
111
112    /// Request cancellation without going through the Sink trait.
113    /// # Errors
114    ///
115    /// Returns an error if the receiving end of the channel has been dropped.
116    pub fn cancel(&self) -> Result<(), mpsc::error::SendError<SyncCommand>> {
117        self.commands.send(SyncCommand::Cancel)
118    }
119
120    /// Clone the cancel sender for external session tracking.
121    #[must_use]
122    pub fn cancel_sender(&self) -> mpsc::UnboundedSender<SyncCommand> {
123        self.commands.clone()
124    }
125
126    #[must_use]
127    pub fn control(&self) -> IntentControl {
128        IntentControl {
129            commands: self.commands.clone(),
130        }
131    }
132}
133
134impl Stream for IntentHandle {
135    type Item = SyncEvent;
136
137    fn poll_next(mut self: Pin<&mut Self>, context: &mut Context<'_>) -> Poll<Option<Self::Item>> {
138        self.events.poll_recv(context)
139    }
140}
141
142impl Sink<SyncCommand> for IntentHandle {
143    type Error = mpsc::error::SendError<SyncCommand>;
144
145    fn poll_ready(self: Pin<&mut Self>, _context: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
146        if self.commands.is_closed() {
147            Poll::Ready(Err(mpsc::error::SendError(SyncCommand::Cancel)))
148        } else {
149            Poll::Ready(Ok(()))
150        }
151    }
152
153    fn start_send(self: Pin<&mut Self>, item: SyncCommand) -> Result<(), Self::Error> {
154        self.commands.send(item)
155    }
156
157    fn poll_flush(self: Pin<&mut Self>, _context: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
158        Poll::Ready(Ok(()))
159    }
160
161    fn poll_close(self: Pin<&mut Self>, _context: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
162        Poll::Ready(Ok(()))
163    }
164}
syncweb_core/sync/intents.rs

syncweb_core/sync/
intents.rs
