modrpc-hub 0.0.1

Multicast transport for modrpc
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

use modrpc::{
    PacketBundle,
    WebSocketIngress,
    WorkerContext,
};
use mproto::BaseLen;

use crate::{
    broadcaster::InPacket,
    BroadcasterHandle,
    TransportIndex,
};

pub async fn spawn_gloo_websocket_spoke(
    worker_context: &WorkerContext,
    broadcaster_handle: BroadcasterHandle,
    buffer_pool: bab::HeapBufferPool,
    websocket: gloo_net::websocket::futures::WebSocket,
    max_packet_size: usize,
) -> TransportIndex {
    use futures_util::StreamExt;

    let (ws_tx, ws_rx) = websocket.split();
    let broadcaster_spoke = broadcaster_handle.add_gloo_ws(Box::new(ws_tx)).await;
    let to_broadcaster = broadcaster_handle.in_packet_sender().clone();

    let mut ingress = WebSocketIngress::new(
        ws_rx,
        buffer_pool.clone(),
        max_packet_size,
    );

    worker_context.spawn(async move {
        while let Ok(packet_bundle) = ingress.receive().await {
            let Ok(header) =
                mproto::decode_value::<PacketBundle>(&packet_bundle[..])
            else {
                continue;
            };

            packet_bundle.advance(PacketBundle::BASE_LEN);

            if let Err(_) =
                to_broadcaster.send(InPacket {
                    transport: broadcaster_spoke,
                    channel_id: header.channel_id,
                    packet: packet_bundle,
                })
                .await
            {
                break;
            }
        }

        broadcaster_handle.remove_transport(broadcaster_spoke).await;
    });

    broadcaster_spoke
}