ferrite_session/internal/session/channel/
send.rs

1use tokio::{
2  task,
3  try_join,
4};
5
6use crate::internal::{
7  base::{
8    unsafe_create_session,
9    unsafe_run_session,
10    AppendContext,
11    Context,
12    ContextLens,
13    Empty,
14    PartialSession,
15    Protocol,
16  },
17  functional::{
18    App,
19    Nat,
20  },
21  protocol::SendChannel,
22};
23
24pub fn send_channel_from<C1, C2, N, A, B>(
25  _n: N,
26  cont: PartialSession<C2, B>,
27) -> PartialSession<C1, SendChannel<A, B>>
28where
29  A: Protocol,
30  B: Protocol,
31  C1: Context,
32  C2: Context,
33  N: ContextLens<C1, A, Empty, Target = C2>,
34{
35  unsafe_create_session::<C1, SendChannel<A, B>, _, _>(
36    move |ctx1, (chan_sender, provider_end)| async move {
37      let (endpoint, ctx2) = N::extract_source(ctx1);
38
39      let client_end = endpoint.get_applied();
40
41      let ctx3 = N::insert_target((), ctx2);
42
43      chan_sender.send(client_end).unwrap();
44
45      unsafe_run_session(cont, ctx3, provider_end).await;
46    },
47  )
48}
49
50pub fn receive_channel_from<C1, C2, C3, N, M, A1, A2, B>(
51  _n: N,
52  cont1: impl FnOnce(M) -> PartialSession<C3, B>,
53) -> PartialSession<C1, B>
54where
55  A1: Protocol,
56  A2: Protocol,
57  B: Protocol,
58  C1: Context<Length = M>,
59  C2: AppendContext<(A1, ()), Appended = C3>,
60  C3: Context,
61  N: ContextLens<C1, SendChannel<A1, A2>, A2, Target = C2>,
62  M: Nat,
63{
64  let cont2 = cont1(M::nat());
65
66  unsafe_create_session(move |ctx1, provider_end| async move {
67    let (endpoint, ctx2) = N::extract_source(ctx1);
68
69    let (chan_receiver, client_end2) = endpoint.get_applied();
70
71    let client_end1 = chan_receiver.recv().await.unwrap();
72
73    let ctx3 = N::insert_target(App::new(client_end2), ctx2);
74
75    let ctx4 = <N::Target as AppendContext<(A1, ())>>::append_context(
76      ctx3,
77      (App::new(client_end1), ()),
78    );
79
80    unsafe_run_session(cont2, ctx4, provider_end).await;
81  })
82}
83
84/*
85   Multiplicative Conjunction, Alternative Parallel Version
86
87
88      cont1 :: Δ ⊢ P    cont2 :: Δ'  ⊢ Q
89   ========================================
90     fork(cont1, cont2) :: Δ, Δ' ⊢ P ⊗ Q
91
92   Takes in two session builders and return a new session builder
93   with its inputs combined and outputs a parallel context
94*/
95
96pub fn fork<A, B, C1, C2>(
97  cont1: PartialSession<C1, A>,
98  cont2: PartialSession<C2, B>,
99) -> PartialSession<C1::Appended, SendChannel<A, B>>
100where
101  A: Protocol,
102  B: Protocol,
103  C1: Context,
104  C2: Context,
105  C1: AppendContext<C2>,
106{
107  unsafe_create_session::<C1::Appended, SendChannel<A, B>, _, _>(
108    move |ctx, (chan_sender, provider_end_b)| async move {
109      let (ctx1, ctx2) = C1::split_context(ctx);
110
111      let (provider_end_a, client_end_a) = A::create_endpoints();
112
113      let child1 = task::spawn(async move {
114        unsafe_run_session(cont1, ctx1, provider_end_a).await;
115      });
116
117      chan_sender.send(client_end_a).unwrap();
118
119      let child2 = task::spawn(async move {
120        unsafe_run_session(cont2, ctx2, provider_end_b).await;
121      });
122
123      try_join!(child1, child2).unwrap();
124    },
125  )
126}
ferrite_session/internal/session/channel/send.rs

ferrite_session/internal/session/channel/
send.rs
