ferrite_session/internal/base/shared/
serialize.rs

1use core::marker::PhantomData;
2
3use tokio::task;
4
5use super::types::*;
6use crate::internal::base::{
7  channel::*,
8  protocol::*,
9};
10
11#[derive(serde::Serialize, serde::Deserialize)]
12pub struct SerializedSharedChannel<S>
13where
14  S: SharedProtocol,
15{
16  acquire_sender: IpcSender<()>,
17  acquire_receiver: IpcReceiver<()>,
18  linear_sender: OpaqueSender,
19  linear_receiver: OpaqueReceiver,
20  phantom: PhantomData<S>,
21}
22
23impl<S> Clone for SerializedSharedChannel<S>
24where
25  S: SharedProtocol,
26{
27  fn clone(&self) -> Self
28  {
29    SerializedSharedChannel {
30      acquire_sender: self.acquire_sender.clone(),
31      acquire_receiver: self.acquire_receiver.clone(),
32      linear_sender: self.linear_sender.clone(),
33      linear_receiver: self.linear_receiver.clone(),
34      phantom: PhantomData,
35    }
36  }
37}
38
39fn serialize_shared_channel<S>(
40  channel: SharedChannel<S>
41) -> SerializedSharedChannel<S>
42where
43  S: SharedProtocol + ForwardChannel,
44{
45  let (sender1, receiver1) = ipc_channel::<()>();
46
47  let (sender2, receiver2) = ipc_channel::<()>();
48
49  let (sender3, receiver3) = opaque_channel();
50
51  let (sender4, receiver4) = opaque_channel();
52
53  task::spawn(async move {
54    loop {
55      let receiver1 = receiver1.clone();
56
57      let signal = task::spawn_blocking(move || receiver1.recv())
58        .await
59        .unwrap();
60
61      match signal {
62        Some(()) => {
63          let (sender5, receiver5) = once_channel::<()>();
64
65          let (sender6, receiver6) = once_channel::<S>();
66
67          {
68            let channel = channel.clone();
69
70            let sender2 = sender2.clone();
71
72            let receiver3 = receiver3.clone();
73
74            let sender4 = sender4.clone();
75
76            debug!("[serialize_shared_channel] acquiring local shared channel");
77
78            channel.endpoint.send((sender5, sender6)).unwrap();
79
80            receiver5.recv().await.unwrap();
81
82            debug!("[serialize_shared_channel] acquired local shared channel");
83
84            sender2.send(());
85
86            receiver6.forward_to(sender4, receiver3);
87          }
88        }
89        None => break,
90      }
91    }
92  });
93
94  SerializedSharedChannel {
95    acquire_sender: sender1,
96    acquire_receiver: receiver2,
97    linear_sender: sender3,
98    linear_receiver: receiver4,
99    phantom: PhantomData,
100  }
101}
102
103fn deserialize_shared_channel<S>(
104  channel: SerializedSharedChannel<S>
105) -> SharedChannel<S>
106where
107  S: SharedProtocol + ForwardChannel + Send,
108{
109  let (sender1, receiver1) = unbounded::<(SenderOnce<()>, SenderOnce<S>)>();
110
111  task::spawn(async move {
112    while let Some((sender2, sender3)) = receiver1.recv().await {
113      debug!("[deserialize_shared_channel] acquiring remote shared channel");
114
115      channel.acquire_sender.send(());
116
117      let acquire_receiver = channel.acquire_receiver.clone();
118
119      task::spawn_blocking(move || {
120        acquire_receiver.recv().unwrap();
121      })
122      .await
123      .unwrap();
124
125      debug!("[deserialize_shared_channel] acquired remote shared channel");
126
127      sender2.send(()).unwrap();
128
129      let channel2 = channel.clone();
130
131      sender3.forward_to(channel2.linear_sender, channel2.linear_receiver);
132    }
133  });
134
135  SharedChannel { endpoint: sender1 }
136}
137
138impl<A> serde::Serialize for SharedChannel<A>
139where
140  A: SharedProtocol + ForwardChannel,
141{
142  fn serialize<S>(
143    &self,
144    serializer: S,
145  ) -> Result<S::Ok, S::Error>
146  where
147    S: serde::Serializer,
148  {
149    serialize_shared_channel(self.clone()).serialize(serializer)
150  }
151}
152
153impl<'a, A> serde::Deserialize<'a> for SharedChannel<A>
154where
155  A: SharedProtocol + ForwardChannel,
156{
157  fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
158  where
159    D: serde::Deserializer<'a>,
160  {
161    let channel = <SerializedSharedChannel<A>>::deserialize(deserializer)?;
162
163    Ok(deserialize_shared_channel(channel))
164  }
165}
ferrite_session/internal/base/shared/serialize.rs

ferrite_session/internal/base/shared/
serialize.rs
