remoc 0.18.3

🦑 Remote multiplexed objects, channels, observable collections and RPC making remote interactions seamless. Provides multiple remote channels and RPC over TCP, TLS or any other transport.
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
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129

#[cfg(feature = "js")]
use wasm_bindgen_test::wasm_bindgen_test;

use crate::loop_channel;

// Avoid imports here to test if proc macro works without imports.

#[derive(Debug, serde::Serialize, serde::Deserialize)]
pub enum IncreaseError {
    Overflow,
    Call(remoc::rtc::CallError),
}

impl From<remoc::rtc::CallError> for IncreaseError {
    fn from(err: remoc::rtc::CallError) -> Self {
        Self::Call(err)
    }
}

pub trait CheckedAddable
where
    Self: Sized,
{
    fn checked_add(self, rhs: Self) -> Option<Self>;
}

impl CheckedAddable for u8 {
    fn checked_add(self, rhs: Self) -> Option<Self> {
        self.checked_add(rhs)
    }
}

#[remoc::rtc::remote]
pub trait GenericCounter<T>
where
    T: remoc::RemoteSend + CheckedAddable + Clone + Default + Sync,
{
    async fn value(&self) -> Result<T, remoc::rtc::CallError>;
    async fn watch(&mut self) -> Result<remoc::rch::watch::Receiver<T>, remoc::rtc::CallError>;
    #[no_cancel]
    async fn increase(&mut self, #[serde(default)] by: T) -> Result<(), IncreaseError>;
}

pub struct GenericCounterObj<T> {
    value: T,
    watchers: Vec<remoc::rch::watch::Sender<T>>,
}

impl<T> GenericCounterObj<T>
where
    T: Default,
{
    pub fn new() -> Self {
        Self { value: T::default(), watchers: Vec::new() }
    }
}

impl<T> GenericCounter<T> for GenericCounterObj<T>
where
    T: remoc::RemoteSend + CheckedAddable + Clone + Default + Sync,
{
    async fn value(&self) -> Result<T, remoc::rtc::CallError> {
        Ok(self.value.clone())
    }

    async fn watch(&mut self) -> Result<remoc::rch::watch::Receiver<T>, remoc::rtc::CallError> {
        let (tx, rx) = remoc::rch::watch::channel(self.value.clone());
        self.watchers.push(tx);
        Ok(rx)
    }

    async fn increase(&mut self, by: T) -> Result<(), IncreaseError> {
        match self.value.clone().checked_add(by) {
            Some(new_value) => self.value = new_value,
            None => return Err(IncreaseError::Overflow),
        }

        for watch in &self.watchers {
            let _ = watch.send(self.value.clone());
        }

        Ok(())
    }
}

#[cfg_attr(not(feature = "js"), tokio::test)]
#[cfg_attr(feature = "js", wasm_bindgen_test)]
async fn simple() {
    use remoc::rtc::Server;

    crate::init();
    let ((mut a_tx, _), (_, mut b_rx)) = loop_channel::<GenericCounterClient<u8>>().await;

    println!("Creating generic counter server");
    let counter_obj = GenericCounterObj::new();
    let (server, client) = GenericCounterServer::new(counter_obj, 1);

    println!("Sending generic counter client");
    a_tx.send(client).await.unwrap();

    let client_task = async move {
        println!("Receiving counter client");
        let mut client = b_rx.recv().await.unwrap().unwrap();

        println!("Spawning watch...");
        let mut watch_rx = client.watch().await.unwrap();
        remoc::exec::spawn(async move {
            while watch_rx.changed().await.is_ok() {
                println!("Watch value: {}", *watch_rx.borrow_and_update().unwrap());
            }
        });

        println!("value: {}", client.value().await.unwrap());
        assert_eq!(client.value().await.unwrap(), 0);

        println!("add 20");
        client.increase(20).await.unwrap();
        println!("value: {}", client.value().await.unwrap());
        assert_eq!(client.value().await.unwrap(), 20);

        println!("add 45");
        client.increase(45).await.unwrap();
        println!("value: {}", client.value().await.unwrap());
        assert_eq!(client.value().await.unwrap(), 65);
    };

    let ((), (_counter_obj, res)) = tokio::join!(client_task, server.serve());
    res.unwrap();
}