wasm-bus 1.1.0

Invocation bus for web assembly modules
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

use derivative::*;
use std::collections::HashMap;
use std::future::Future;
use std::pin::Pin;
use std::sync::Arc;
use std::sync::Mutex;
#[allow(unused_imports, dead_code)]
use tracing::{debug, error, info, trace, warn};
use wasm_bus_types::SerializationFormat;

use crate::abi::CallError;
use crate::abi::CallHandle;

type CallbackHandler = Arc<
    dyn Fn(CallHandle, Vec<u8>) -> Pin<Box<dyn Future<Output = Result<Vec<u8>, CallError>> + Send>>
        + Send
        + Sync,
>;

#[derive(Derivative, Clone)]
#[derivative(Debug)]
pub struct RespondToService {
    pub(crate) format: SerializationFormat,
    #[derivative(Debug = "ignore")]
    pub(crate) callbacks: Arc<Mutex<HashMap<CallHandle, CallbackHandler>>>,
    pub(crate) persistent: bool,
}

impl RespondToService {
    pub fn new(format: SerializationFormat, persistent: bool) -> RespondToService {
        RespondToService {
            format,
            callbacks: Default::default(),
            persistent,
        }
    }

    pub fn add(
        &self,
        handle: CallHandle,
        callback: Arc<
            dyn Fn(
                    CallHandle,
                    Vec<u8>,
                )
                    -> Pin<Box<dyn Future<Output = Result<Vec<u8>, CallError>> + Send>>
                + Send
                + Sync,
        >,
    ) {
        let mut callbacks = self.callbacks.lock().unwrap();
        callbacks.insert(handle, callback);
    }

    pub fn remove(&self, handle: &CallHandle) -> Option<CallbackHandler> {
        let mut callbacks = self.callbacks.lock().unwrap();
        callbacks.remove(handle)
    }

    pub async fn process(&self, callback_handle: CallHandle, handle: CallHandle, request: Vec<u8>) {
        let callback = {
            let callbacks = self.callbacks.lock().unwrap();
            if let Some(callback) = callbacks.get(&callback_handle) {
                Arc::clone(callback)
            } else {
                let err: u32 = CallError::InvalidHandle.into();
                crate::abi::syscall::fault(handle, err as u32);
                crate::engine::BusEngine::remove(&handle, "invalid callback handle");
                return;
            }
        };

        let res = callback.as_ref()(handle, request);
        match res.await {
            Ok(a) => {
                crate::abi::syscall::reply(handle, &a[..]);
            }
            Err(err) => {
                let err: u32 = err.into();
                crate::abi::syscall::fault(handle, err as u32);
            }
        }
        if self.persistent == false {
            crate::engine::BusEngine::remove(&handle, "request was processed (by respond_to)");
        }
    }
}