ferrite_session/internal/session/
run.rs

1use tokio::{
2  task,
3  try_join,
4};
5
6use crate::internal::{
7  base::{
8    once_channel,
9    unbounded,
10    unsafe_create_shared_channel,
11    unsafe_run_session,
12    unsafe_run_shared_session,
13    Protocol,
14    Session,
15    SharedChannel,
16    SharedProtocol,
17    SharedSession,
18    Value,
19  },
20  protocol::{
21    End,
22    SendValue,
23  },
24};
25
26pub async fn run_session(session: Session<End>)
27{
28  let (sender, receiver) = once_channel();
29
30  let child1 = task::spawn(async move {
31    unsafe_run_session(session, (), sender).await;
32  });
33
34  let child2 = task::spawn(async move {
35    receiver.recv().await.unwrap();
36  });
37
38  try_join!(child1, child2).unwrap();
39}
40
41pub async fn run_session_with_result<T>(
42  session: Session<SendValue<T, End>>
43) -> T
44where
45  T: Send + 'static,
46{
47  let (provider_end, val_receiver) = <SendValue<T, End>>::create_endpoints();
48
49  let child1 = task::spawn(async move {
50    unsafe_run_session(session, (), provider_end).await;
51  });
52
53  let (Value(val), end_receiver) = val_receiver.recv().await.unwrap();
54
55  end_receiver.recv().await.unwrap();
56
57  let _ = child1.await;
58
59  val
60}
61
62pub fn run_shared_session<A>(session: SharedSession<A>) -> SharedChannel<A>
63where
64  A: SharedProtocol,
65{
66  let (chan, _) = run_shared_session_with_join_handle(session);
67
68  chan
69}
70
71pub fn run_shared_session_with_join_handle<A>(
72  session: SharedSession<A>
73) -> (SharedChannel<A>, task::JoinHandle<()>)
74where
75  A: SharedProtocol,
76{
77  let (sender1, receiver1) = unbounded();
78
79  let (session2, receiver2) = unsafe_create_shared_channel();
80
81  task::spawn(async move {
82    info!("[run_shared_session] exec_shared_session");
83
84    unsafe_run_shared_session(session, receiver1).await;
85
86    info!("[run_shared_session] exec_shared_session returned");
87  });
88
89  let handle = task::spawn(async move {
90    loop {
91      let m_senders = receiver2.recv().await;
92
93      debug!("[run_shared_session] received sender3");
94
95      match m_senders {
96        Some(senders) => {
97          sender1.send(senders).unwrap();
98        }
99        None => {
100          info!("[run_shared_session] terminating shared session");
101
102          return;
103        }
104      }
105    }
106  });
107
108  (session2, handle)
109}
ferrite_session/internal/session/run.rs

ferrite_session/internal/session/
run.rs
