swarm-discovery 0.6.3

Discovery service for IP-based swarms
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
130
131
132
133
134

use crate::{
    receiver::{receiver, ReceiverError},
    sender::{self, sender},
    socket::Sockets,
    updater::updater,
    Discoverer,
};
use acto::{AcTokioRuntime, ActoCell, ActoInput, ActoRef};
use hickory_proto::rr::Name;
use std::{collections::HashMap, mem::replace, net::IpAddr};

pub enum Input {
    RemoveAll,
    RemovePort(u16),
    RemoveAddr(IpAddr),
    AddAddr(u16, Vec<IpAddr>),
    SetTxt(String, Option<String>),
    RemoveTxt(String),
    AddInterface(IpAddr),
    RemoveInterface(IpAddr),
}

pub async fn guardian(
    mut ctx: ActoCell<Input, AcTokioRuntime, Result<(), ReceiverError>>,
    mut discoverer: Discoverer,
    sockets: Sockets,
    service_name: Name,
) {
    let callback = replace(&mut discoverer.callback, Box::new(|_, _| {}));
    let tau = discoverer.tau;
    let phi = discoverer.phi;
    let upd_ref = ctx.supervise(
        ctx.spawn("updater", move |ctx| updater(ctx, tau, phi, callback))
            .map_handle(Ok),
    );

    let sockets2 = sockets.clone();
    let sn = service_name.clone();
    let snd_ref = ctx.supervise(
        ctx.spawn("sender", move |ctx| {
            sender(ctx, sockets, upd_ref, discoverer, sn)
        })
        .map_handle(Ok),
    );

    if let Some(v4) = sockets2.v4() {
        let service_name = service_name.clone();
        let snd_ref = snd_ref.clone();
        ctx.spawn_supervised("receiver_v4", move |ctx| {
            receiver(ctx, service_name, v4, snd_ref)
        });
    }

    if let Some(v6) = sockets2.v6() {
        let service_name = service_name.clone();
        let snd_ref = snd_ref.clone();
        ctx.spawn_supervised("receiver_v6", move |ctx| {
            receiver(ctx, service_name, v6, snd_ref)
        });
    }

    // Track interface receivers so we can stop them when interfaces are removed
    let mut interface_receivers: HashMap<IpAddr, ActoRef<()>> = HashMap::new();

    // Start receivers for initial interface sockets
    let initial_interfaces = sockets2.get_all_interface_addresses_v4();
    for addr in initial_interfaces {
        if let Some(socket) = sockets2.get_interface_socket_v4(addr) {
            let service_name = service_name.clone();
            let snd_ref = snd_ref.clone();
            let addr_str = addr.to_string();
            let receiver_ref = ctx
                .spawn_supervised(&format!("receiver_interface_{}", addr_str), move |ctx| {
                    receiver(ctx, service_name, socket, snd_ref)
                });
            interface_receivers.insert(IpAddr::V4(addr), receiver_ref);
            tracing::info!("Started receiver for initial interface {}", addr);
        }
    }

    // only stop when a supervised actor stops
    loop {
        let msg = ctx.recv().await;
        match msg {
            ActoInput::NoMoreSenders => {}
            ActoInput::Supervision { id, name, result } => {
                match result {
                    Ok(Ok(_)) => tracing::warn!("actor {:?} ({}) stopped", id, name),
                    Ok(Err(e)) => {
                        tracing::warn!("actor {:?} ({}) failed: {}", id, name, e)
                    }
                    Err(e) => {
                        tracing::warn!("actor {:?} ({}) aborted: {}", id, name, e);
                    }
                }
                break;
            }
            ActoInput::Message(msg) => match &msg {
                Input::AddInterface(addr) => {
                    if let IpAddr::V4(ipv4) = addr {
                        if let Err(e) = sockets2.add_interface_v4(*ipv4) {
                            tracing::warn!("Failed to add interface {}: {}", addr, e);
                        } else {
                            // Start a receiver for the new interface socket
                            if let Some(socket) = sockets2.get_interface_socket_v4(*ipv4) {
                                let service_name = service_name.clone();
                                let snd_ref = snd_ref.clone();
                                let addr_str = addr.to_string();
                                let receiver_ref = ctx.spawn_supervised(
                                    &format!("receiver_interface_{}", addr_str),
                                    move |ctx| receiver(ctx, service_name, socket, snd_ref),
                                );
                                interface_receivers.insert(*addr, receiver_ref);
                                tracing::info!("Started receiver for interface {}", addr);
                            }
                        }
                    }
                }
                Input::RemoveInterface(addr) => {
                    if let IpAddr::V4(ipv4) = addr {
                        sockets2.remove_interface_v4(*ipv4);
                        // Remove the receiver reference for this interface
                        if interface_receivers.remove(addr).is_some() {
                            tracing::info!("Removed receiver reference for interface {}", addr);
                        }
                    }
                }
                _ => {
                    snd_ref.send(sender::MdnsMsg::Update(msg));
                }
            },
        }
    }
}