Add pre-built tasks

.gitignore

@@ -6,3 +6,4 @@
 
 # direnv (https://direnv.net/)
 .envrc
+.direnv/

Cargo.toml

@@ -10,16 +10,16 @@ expose_stream = []
 [dependencies]
 bounded-static = "0.5.0"
 bytes = "1.5.0"
-imap-codec = { version = "2.0.0", features = ["quirk_crlf_relaxed", "bounded-static"] }
-imap-types = { version = "2.0.0" }
+imap-codec = { version = "2.0.0", features = ["quirk_crlf_relaxed", "bounded-static", "ext_id", "ext_sort_thread", "ext_binary", "ext_uidplus"] }
+imap-types = { version = "2.0.0", features = ["ext_id", "ext_sort_thread", "ext_binary", "ext_uidplus"] }
 thiserror = "1.0.49"
-tokio = { version = "1.32.0", features = ["io-util"] }
+tokio = { version = "1.32.0", features = ["io-util", "sync", "time"] }
 tracing = "0.1.40"
 
 [dev-dependencies]
 rand = "0.8.5"
 tag-generator = { path = "tag-generator" }
-tokio = { version = "1.32.0", features = ["macros", "net", "rt", "sync"] }
+tokio = { version = "1.32.0", features = ["full"] }
 
 [workspace]
 resolver = "2"

src/client.rs

@@ -1,4 +1,7 @@
-use std::fmt::{Debug, Formatter};
+use std::{
+    fmt::{Debug, Formatter},
+    time::Duration,
+};
 
 use bytes::BytesMut;
 use imap_codec::{
@@ -12,6 +15,7 @@ use imap_types::{
     secret::Secret,
 };
 use thiserror::Error;
+use tokio::sync::mpsc;
 
 use crate::{
     handle::{Handle, HandleGenerator, HandleGeneratorGenerator, RawHandle},
@@ -24,6 +28,12 @@ use crate::{
 static HANDLE_GENERATOR_GENERATOR: HandleGeneratorGenerator<ClientFlowCommandHandle> =
     HandleGeneratorGenerator::new();
 
+/// The default IDLE timeout, in seconds, as defined in RFC2177.
+///
+/// > Clients using IDLE are advised to terminate the IDLE and
+/// re-issue it at least every 29 minutes to avoid being logged off.
+static DEFAULT_IDLE_TIMEOUT: u64 = 60 * 29;
+
 #[derive(Clone, Debug, PartialEq)]
 #[non_exhaustive]
 pub struct ClientFlowOptions {
@@ -45,12 +55,14 @@ pub struct ClientFlow {
     handle_generator: HandleGenerator<ClientFlowCommandHandle>,
     send_command_state: SendCommandState,
     receive_response_state: ReceiveState<ResponseCodec>,
+    idle_timeout: Option<u64>,
 }
 
 impl Debug for ClientFlow {
     fn fmt(&self, f: &mut Formatter) -> std::fmt::Result {
         f.debug_struct("ClientFlow")
             .field("handle_generator", &self.handle_generator)
+            .field("idle_timeout", &self.idle_timeout)
             .finish_non_exhaustive()
     }
 }
@@ -111,11 +123,24 @@ impl ClientFlow {
             handle_generator: HANDLE_GENERATOR_GENERATOR.generate(),
             send_command_state,
             receive_response_state,
+            idle_timeout: None,
         };
 
         Ok((client_flow, greeting))
     }
 
+    pub fn set_some_idle_timeout(&mut self, secs: Option<u64>) {
+        self.idle_timeout = secs;
+    }
+
+    pub fn set_idle_timeout(&mut self, secs: u64) {
+        self.set_some_idle_timeout(Some(secs));
+    }
+
+    pub fn get_idle_timeout(&self) -> Duration {
+        Duration::from_secs(self.idle_timeout.unwrap_or(DEFAULT_IDLE_TIMEOUT))
+    }
+
     /// Enqueues the [`Command`] for being sent to the client.
     ///
     /// The [`Command`] is not sent immediately but during one of the next calls of
@@ -152,8 +177,22 @@ impl ClientFlow {
                 return Ok(event);
             }
 
-            if let Some(event) = self.progress_receive().await? {
-                return Ok(event);
+            if self.is_waiting_for_idle_done_set() {
+                let timeout =
+                    tokio::time::timeout(self.get_idle_timeout(), self.progress_receive());
+                match timeout.await {
+                    Ok(Ok(Some(event))) => break Ok(event),
+                    Ok(Ok(None)) => continue,
+                    Ok(Err(err)) => break Err(err),
+                    Err(_) => {
+                        self.set_idle_done();
+                        continue;
+                    }
+                }
+            } else {
+                if let Some(event) = self.progress_receive().await? {
+                    return Ok(event);
+                }
             }
         }
     }
@@ -272,7 +311,7 @@ impl ClientFlow {
     pub fn set_authenticate_data(
         &mut self,
         authenticate_data: AuthenticateData<'static>,
-    ) -> Result<ClientFlowCommandHandle, AuthenticateData> {
+    ) -> Result<ClientFlowCommandHandle, AuthenticateData<'static>> {
         self.send_command_state
             .set_authenticate_data(authenticate_data)
     }
@@ -281,6 +320,10 @@ impl ClientFlow {
         self.send_command_state.set_idle_done()
     }
 
+    pub fn is_waiting_for_idle_done_set(&self) -> bool {
+        self.send_command_state.is_waiting_for_idle_done_set()
+    }
+
     #[cfg(feature = "expose_stream")]
     /// Return the underlying stream for debug purposes (or experiments).
     ///
@@ -393,6 +436,8 @@ pub enum ClientFlowEvent {
 pub enum ClientFlowError {
     #[error("Stream was closed")]
     StreamClosed,
+    #[error("Queue receiver was closed")]
+    QueueClosed(#[source] mpsc::error::SendError<()>),
     #[error(transparent)]
     Io(#[from] tokio::io::Error),
     #[error("Expected `\\r\\n`, got `\\n`")]

src/send_command.rs

@@ -19,6 +19,7 @@ use crate::{
     types::CommandAuthenticate,
 };
 
+#[derive(Debug)]
 pub struct SendCommandState {
     command_codec: CommandCodec,
     authenticate_data_codec: AuthenticateDataCodec,
@@ -51,6 +52,16 @@ impl SendCommandState {
         }
     }
 
+    pub fn is_waiting_for_idle_done_set(&self) -> bool {
+        matches!(
+            self.current_command,
+            Some(CurrentCommand::Idle(IdleState {
+                activity: IdleActivity::WaitingForIdleDoneSet,
+                ..
+            }))
+        )
+    }
+
     pub fn enqueue(&mut self, handle: ClientFlowCommandHandle, command: Command<'static>) {
         self.queued_commands
             .push_back(QueuedCommand { handle, command });
@@ -191,7 +202,7 @@ impl SendCommandState {
     pub fn set_authenticate_data(
         &mut self,
         authenticate_data: AuthenticateData<'static>,
-    ) -> Result<ClientFlowCommandHandle, AuthenticateData> {
+    ) -> Result<ClientFlowCommandHandle, AuthenticateData<'static>> {
         // Check whether in correct state
         let Some(current_command) = self.current_command.take() else {
             return Err(authenticate_data);
@@ -336,6 +347,7 @@ impl SendCommandState {
 }
 
 /// Queued (and not sent yet) command.
+#[derive(Debug)]
 struct QueuedCommand {
     handle: ClientFlowCommandHandle,
     command: Command<'static>,
@@ -396,6 +408,7 @@ impl QueuedCommand {
 }
 
 /// Currently being sent command.
+#[derive(Debug)]
 enum CurrentCommand {
     /// Sending state of regular command.
     Command(CommandState),
@@ -453,6 +466,7 @@ impl<S> FinishSendingResult<S> {
     }
 }
 
+#[derive(Debug)]
 struct CommandState {
     handle: ClientFlowCommandHandle,
     command: Command<'static>,
@@ -533,6 +547,7 @@ impl CommandState {
     }
 }
 
+#[derive(Debug)]
 enum CommandActivity {
     /// Pushing fragments to the write buffer.
     PushingFragments {
@@ -552,6 +567,7 @@ enum CommandActivity {
     },
 }
 
+#[derive(Debug)]
 struct AuthenticateState {
     handle: ClientFlowCommandHandle,
     command_authenticate: CommandAuthenticate,
@@ -603,6 +619,7 @@ impl AuthenticateState {
     }
 }
 
+#[derive(Debug)]
 enum AuthenticateActivity {
     /// Pushing the authenticate command to the write buffer.
     PushingAuthenticate { authenticate: Vec<u8> },
@@ -621,6 +638,7 @@ enum AuthenticateActivity {
     WaitingForAuthenticateDataSent,
 }
 
+#[derive(Debug)]
 struct IdleState {
     handle: ClientFlowCommandHandle,
     tag: Tag<'static>,
@@ -668,6 +686,7 @@ impl IdleState {
     }
 }
 
+#[derive(Debug)]
 enum IdleActivity {
     /// Pushing the idle command to the write buffer.
     PushingIdle { idle: Vec<u8> },

src/stream.rs

@@ -127,7 +127,7 @@ pub struct ReadBuffer {
 }
 
 /// Buffer for writing bytes with [`AnyStream::write`].
-#[derive(Default)]
+#[derive(Debug, Default)]
 pub struct WriteBuffer {
     pub bytes: BytesMut,
 }

tasks/Cargo.toml

@@ -6,7 +6,8 @@ license = "MIT OR Apache-2.0"
 
 [dependencies]
 imap-flow = { path = ".." }
-imap-types = "2.0.0"
+imap-types = { version = "2.0.0", features = ["ext_id", "ext_sort_thread", "ext_binary", "ext_uidplus"] }
 tag-generator = { path = "../tag-generator" }
 thiserror = "1.0.50"
 tokio = { version = "1.32.0", features = ["full"] }
+tracing = "0.1"

tasks/examples/client-tasks.rs

@@ -1,26 +1,18 @@
-use imap_flow::{
-    client::{ClientFlow, ClientFlowOptions},
-    stream::AnyStream,
-};
+use imap_flow::{client::ClientFlowOptions, stream::AnyStream};
 use imap_types::response::{Response, Status};
 use tasks::{
-    tasks::{AuthenticatePlainTask, CapabilityTask, LogoutTask},
+    tasks::{authenticate::AuthenticateTask, capability::CapabilityTask, logout::LogoutTask},
     Scheduler, SchedulerEvent,
 };
 use tokio::net::TcpStream;
 
 #[tokio::main]
 async fn main() {
     let mut scheduler = {
-        let (flow, _) = {
-            let stream = TcpStream::connect("127.0.0.1:12345").await.unwrap();
-
-            ClientFlow::receive_greeting(AnyStream::new(stream), ClientFlowOptions::default())
-                .await
-                .unwrap()
-        };
-
-        Scheduler::new(flow)
+        let stream = TcpStream::connect("127.0.0.1:12345").await.unwrap();
+        Scheduler::try_new_flow(AnyStream::new(stream), ClientFlowOptions::default())
+            .await
+            .unwrap()
     };
 
     let handle1 = scheduler.enqueue_task(CapabilityTask::default());
@@ -43,7 +35,7 @@ async fn main() {
         }
     }
 
-    let handle2 = scheduler.enqueue_task(AuthenticatePlainTask::new("alice", "pa²²w0rd", true));
+    let handle2 = scheduler.enqueue_task(AuthenticateTask::new_plain("alice", "pa²²w0rd", true));
     let handle3 = scheduler.enqueue_task(LogoutTask::default());
 
     loop {