refactor: Implement sans I/O

[jakoschiko]

These types are now sans I/O:

• imap_flow::client::ClientFlow
• imap_flow::server::ServerFlow
• imap_tasks::Scheduler

All of them implement the new trait imap_flow::Flow which provides just enough interface for implementing I/O utilities.

For interacting with I/O there is now the optional utility imap_flow::stream::Stream based on tokio and tokio_rustls. This utility and its dependencies can be enabled/disabled with the feature flag stream which is enabled by default.

The current usage looks like this:

let stream = TcpStream::connect("127.0.0.1:12345").await.unwrap();
let mut stream = Stream::insecure(stream);
let mut client = ClientFlow::new(ClientFlowOptions::default());

let greeting = loop {
    match stream.progress(&mut client).await.unwrap() {
        ClientFlowEvent::GreetingReceived { greeting } => break greeting,
        event => println!("unexpected event: {event:?}"),
    }
};

This PR is only the first (big) step. I think there are a lot of smaller API changes and follow-up refactorings that we can and should do. But for now let's focus on the rough design.

Closes #154
Closes #37
Closes #134
Relates to #98

[coveralls]

Pull Request Test Coverage Report for Build 8974528434

Details

• 302 of 556 (54.32%) changed or added relevant lines in 8 files are covered.
• 6 unchanged lines in 3 files lost coverage.
• Overall coverage decreased (-1.4%) to 52.662%

---

Changes Missing Coverage	Covered Lines	Changed/Added Lines	%
src/send_command.rs	25	30	83.33%
src/receive.rs	39	62	62.9%
src/client.rs	72	132	54.55%
src/server.rs	44	120	36.67%
src/stream.rs	59	149	39.6%

Files with Coverage Reduction	New Missed Lines	%
src/receive.rs	1	69.66%
src/server.rs	1	35.17%
src/stream.rs	4	39.1%

Totals
Change from base Build 8951834476:	-1.4%
Covered Lines:	732
Relevant Lines:	1390

---

💛 - Coveralls

[soywod]

So neat! Looks awesome 🙂

[duesee]

Thanks, Jakob! I really like where this is going! 🎉

examples, flow-test, proxy, and tasks look very good to me. src is rather difficult to review. But: This was expected as it's a large refactoring. We also discussed about many changes before.

I would like to put the proxy and the examples under test tomorrow. Also, I'd like to have a better understanding of the TLS code. Maybe it makes sense to discuss this offline.

So far, this looks all very promising :-)

[soywod]

I started to play a bit with the new SansIO API within Himalaya directly, I think I understood well the concept and it's awesome. Yet I have a question, let's take the following code from your example:

let stream = TcpStream::connect("127.0.0.1:12345").await.unwrap();
let mut stream = Stream::insecure(stream);
let client = ClientFlow::new(ClientFlowOptions::default());
let mut scheduler = Scheduler::new(client);

let handle1 = scheduler.enqueue_task(CapabilityTask::default());

loop {
    match stream.progress(&mut scheduler).await.unwrap() {
        SchedulerEvent::TaskFinished(mut token) => {
            if let Some(capability) = handle1.resolve(&mut token) {
                println!("handle1: {capability:?}");
                break;
            }
        }
        SchedulerEvent::Unsolicited(unsolicited) => {
            println!("unsolicited: {unsolicited:?}");

            if let Response::Status(Status::Bye { .. }) = unsolicited {
                break;
            }
        }
    }
}

The scheduler takes care of enqueueing the task and processing the event, and the client (the lib consumer) needs to make the flow progress by running stream.progress.

The problem: the previous code needs to be written for every single task, and it needs to be done client side because the scheduler does not know anymore how to progress (async). Only stream knows, but stream does not know how to process a task or an event.

I see that a 3rd actor could be beneficial here, to prevent lib consumers to write their own helpers. If a lib consumer uses the scheduler and the stream (tokio + rustls), then the lib should be able to provide those helpers. But where to put it and how to name it?

Just to illustrate what I try to say:

let stream = TcpStream::connect("127.0.0.1:12345").await.unwrap();
let stream = Stream::insecure(stream);
let client = ClientFlow::new(ClientFlowOptions::default());
let scheduler = Scheduler::new(client);

// this is the 3rd actor I was thinking about
let mut session = Session::new(stream, scheduler)

let capabilities = session.capability().await.unwrap();

[jakoschiko]

@soywod: I don't see how the situation has changed with sans I/O. But I would agree that we need better abstractions. Here another proposal:

let stream = TcpStream::connect("127.0.0.1:12345").await.unwrap();
let stream = Stream::insecure(stream);
let client = ClientFlow::new(ClientFlowOptions::default());
let scheduler = Scheduler::new(client);

// Session is also sans I/O
let mut session = Session::new(scheduler);

// Borrows &mut session
let mut capability_request = session.capability();

let capabilities = stream.progress(&mut capability_request).await.unwrap();

[soywod]

@soywod: I don't see how the situation has changed with sans I/O.

In my initial PR, I added a exec_task helper directly inside Scheduler, but it is not possible anymore since now the progress needs to be done via Stream.

But I would agree that we need better abstractions.

It is not really a better abstraction, more like a missing one, because I really like the existing you are proposing here.

Here another proposal: session is also sans I/O

Alright, I will give it a shot, thank you!

[jakoschiko]

@soywod: If you add another method to Stream then it would look very similar to your original suggestion:

// Session is also sans I/O
let mut session = Session::new(scheduler);

// Stream::complete takes ownership of the given Flow
let capabilities = stream.complete(session.capability()).await.unwrap();

[soywod]

Stream::complete takes ownership of the given Flow

I see what you mean, but by taking the ownership of the Flow, you also take ownership of the current scheduler which prevents you to run multiple complete in a row. I'm experimenting around but cannot find yet sth satisfying.

[jakoschiko]

No, only the ownership of the result of session.capability() which borrows &mut session. But for making this work we would need to remove the T from Stream<T> which is only a phantom type anyway. Should I remove the type parameter in this PR?

[soywod]

No, only the ownership of the result of session.capability() which borrows &mut session.

But how to implement then the Flow::enqueue_input? It requires a mutable ref to the scheduler. I can make it work but with a Mutex, which I'm not fan of:

pub struct TaskResolver<T: Task> {
    scheduler: Arc<Mutex<Scheduler>>,
    handle: TaskHandle<T>,
}

impl<T: Task> TaskResolver<T> {
    pub fn new(scheduler: Arc<Mutex<Scheduler>>, task: T) -> Self {
        let handle = scheduler.lock().unwrap().enqueue_task(task);
        Self { scheduler, handle }
    }
}

impl<T: Task> Flow for TaskResolver<T> {
    type Event = T::Output;
    type Error = SchedulerError;

    fn enqueue_input(&mut self, bytes: &[u8]) {
        self.scheduler.lock().unwrap().enqueue_input(bytes);
    }

    fn progress(&mut self) -> Result<Self::Event, FlowInterrupt<Self::Error>> {
        loop {
            match self.scheduler.lock().unwrap().progress()? {
                SchedulerEvent::TaskFinished(mut token) => {
                    if let Some(output) = self.handle.resolve(&mut token) {
                        break Ok(output);
                    }
                }
                SchedulerEvent::Unsolicited(unsolicited) => {
                    if let Response::Status(Status::Bye(bye)) = unsolicited {
                        break Err(SchedulerError::UnexpectedByeResponse(bye).into());
                    } else {
                        println!("unsolicited: {unsolicited:?}");
                    }
                }
            }
        }
    }
}

Which is then used this way:

pub struct Resolver {
    scheduler: Arc<Mutex<Scheduler>>,
}

impl Resolver {
    pub fn capability(&mut self) -> TaskResolver<CapabilityTask> {
        TaskResolver::new(self.scheduler.clone(), CapabilityTask::new())
    }
}

And allows sth like:

let scheduler = Scheduler::new(client);
let mut resolver = Resolver::new(scheduler);

let capabilities = stream.consume(resolver.capability()).await.unwrap();
let auth = stream
        .consume(resolver.authenticate_plain("a", "b"))
        .await
        .unwrap();

But for making this work we would need to remove the T from Stream<T> which is only a phantom type anyway. Should I remove the type parameter in this PR?

Yes indeed, the T prevents us to execute multiple action with a different Flow event. I tried locally and seems to fix this issue.

[soywod]

Wait I have a better solution, I will come back to you this afternoon.

[jakoschiko]

Hm, maybe I didn't understand your use-case correctly. Do you want to process tasks sequentially or in parallel?

I committed some small changes. Stream has no type parameter anymore and Stream::progress takes ownership of the Flow:

pub async fn progress<F: Flow>(
    &mut self,
    mut flow: F,
) -> Result<F::Event, StreamError<F::Error>>

And I added this implementation:

impl<F: Flow> Flow for &mut F

Now you can decide whether you move or borrow the Flow:

// Borrows flow
stream.progress(&mut flow).await.unwrap();

// Moves flow
stream.progress(flow).await.unwrap();

This should give you a lot of flexibility. Let's discuss your Session idea later, I need some rest now. I hope we can merge this PR soon.

[jakoschiko]

I merge it now. Thank you all.

[duesee]

Thank you all! Awesome team effort :-)