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
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75

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

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

pub async fn spawn_tcp_spoke(
    // TODO take a spawner instead of a full context
    worker_context: &WorkerContext,
    broadcaster_handle: BroadcasterHandle,
    buffer_pool: bab::HeapBufferPool,
    stream: tokio::net::TcpStream,
    max_packet_size: usize,
) -> (TransportIndex, bab::SignalTree) {
    let (tcp_read, tcp_write) = stream.into_split();

    let broadcaster_spoke = broadcaster_handle.add_tcp(tcp_write).await;
    let to_broadcaster = broadcaster_handle.in_packet_sender().clone();
    let shutdown_signal = bab::SignalTree::new();

    let mut ingress = TcpIngress::new(
        tcp_read,
        buffer_pool,
        max_packet_size,
    );

    worker_context.spawn(probius::enter_component_async(
        "tcp-ingress-task", {
            let shutdown_signal = shutdown_signal.clone();
            async move {
                let tracer = probius::new_trace_source("loop");

                while let Ok(packet_bundle) = ingress.receive().await {
                    let result: Result<(), localq::mpsc::SendError<_>> = tracer.trace_future(async {
                        let Ok(header) =
                            mproto::decode_value::<PacketBundle>(&packet_bundle[..])
                        else {
                            return Ok(());
                        };
                        packet_bundle.advance(PacketBundle::BASE_LEN);

                        tracer.trace(|| {
                            probius::trace_metric("bundle_payload_size", packet_bundle.len() as i64);
                        });

                        to_broadcaster.send(InPacket {
                            transport: broadcaster_spoke,
                            channel_id: header.channel_id,
                            packet: packet_bundle,
                        })
                        .await?;

                        Ok(())
                    })
                    .await;
                    if let Err(_) = result {
                        break;
                    }
                }

                shutdown_signal.notify();
                broadcaster_handle.remove_transport(broadcaster_spoke).await;
            }
        },
    ));

    (broadcaster_spoke, shutdown_signal)
}