imsg-session 0.1.0

Session lifecycle — SDP lookup, RFCOMM connect, OBEX handshake, reconnect, and MNS relay
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39

//! Shared accept-loop driver used by MNS session and relay modules.

use std::future::Future;

use futures::StreamExt;
use tokio::io::{AsyncRead, AsyncWrite};
use tokio::sync::watch;

/// `drain` returns `false` to halt early, `true` to accept next.
pub(crate) async fn run_accept_loop<S, T, F, Fut>(
    mut stream_source: S,
    mut cancel: watch::Receiver<bool>,
    mut drain: F,
) where
    S: futures::Stream<Item = T> + Unpin,
    T: AsyncRead + AsyncWrite + Unpin,
    F: FnMut(T, watch::Receiver<bool>) -> Fut,
    Fut: Future<Output = bool>,
{
    loop {
        if *cancel.borrow() {
            return;
        }
        let stream = tokio::select! {
            biased;
            result = cancel.changed() => {
                if result.is_err() || *cancel.borrow_and_update() { return; }
                continue;
            }
            maybe = stream_source.next() => match maybe {
                Some(s) => s,
                None => return,
            }
        };
        if !drain(stream, cancel.clone()).await {
            return;
        }
    }
}