gadget-sdk 0.6.2

SDK for building Blueprints and gadget on Tangle Network
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

#![allow(unused_results)]

use crate::network::gossip::{GossipMessage, NetworkService};

use crate::{error, trace};
use libp2p::gossipsub::TopicHash;
use libp2p::{gossipsub, PeerId};
use std::sync::atomic::AtomicU32;
use std::sync::Arc;

impl NetworkService<'_> {
    #[tracing::instrument(skip(self, event))]
    pub(crate) async fn handle_gossip(&mut self, event: gossipsub::Event) {
        use gossipsub::Event::{GossipsubNotSupported, Message, Subscribed, Unsubscribed};
        let with_connected_peers = |topic: &TopicHash, f: fn(&Arc<AtomicU32>)| {
            let maybe_mapping = self
                .inbound_mapping
                .iter()
                .find(|r| r.0.to_string() == topic.to_string());
            match maybe_mapping {
                Some((_, _, connected_peers)) => {
                    f(connected_peers);
                    true
                }
                None => false,
            }
        };
        match event {
            Message {
                propagation_source,
                message_id,
                message,
            } => {
                self.handle_gossip_message(propagation_source, message_id, message)
                    .await;
            }
            Subscribed { peer_id, topic } => {
                let added = with_connected_peers(&topic, |connected_peers| {
                    connected_peers.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
                });
                if added {
                    trace!("{peer_id} subscribed to {topic}");
                } else {
                    error!("{peer_id} subscribed to unknown topic: {topic}");
                }
            }
            Unsubscribed { peer_id, topic } => {
                let removed = with_connected_peers(&topic, |connected_peers| {
                    connected_peers.fetch_sub(1, std::sync::atomic::Ordering::Relaxed);
                });
                if removed {
                    trace!("{peer_id} unsubscribed from {topic}");
                } else {
                    error!("{peer_id} unsubscribed from unknown topic: {topic}");
                }
            }
            GossipsubNotSupported { peer_id } => {
                trace!("{peer_id} does not support gossipsub!");
            }
        }
    }

    #[tracing::instrument(
    skip(self, message),
    fields(
    %_message_id,
    %_propagation_source,
    source = ?message.source
    )
    )]
    async fn handle_gossip_message(
        &mut self,
        _propagation_source: PeerId,
        _message_id: gossipsub::MessageId,
        message: gossipsub::Message,
    ) {
        let Some(origin) = message.source else {
            error!("Got message from unknown peer");
            return;
        };

        // Reject messages from self
        if origin == self.my_id {
            return;
        }

        trace!("Got message from peer: {origin}");
        match bincode::deserialize::<GossipMessage>(&message.data) {
            Ok(GossipMessage { topic, raw_payload }) => {
                if let Some((_, tx, _)) = self
                    .inbound_mapping
                    .iter()
                    .find(|r| r.0.to_string() == topic)
                {
                    if let Err(e) = tx.send(raw_payload) {
                        error!("Failed to send message to worker: {e}");
                    }
                } else {
                    error!("No registered worker for topic: {topic}!");
                }
            }
            Err(e) => {
                error!("Failed to deserialize message: {e}");
            }
        }
    }
}