connexa/task/
floodsub.rs

1use crate::behaviour::peer_store::store::Store;
2use crate::error::{Error, Protocol, floodsub};
3use crate::prelude::{FloodsubMessage, PubsubFloodsubPublish};
4use crate::task::ConnexaTask;
5use crate::types::{FloodsubCommand, FloodsubEvent};
6use futures::channel::mpsc;
7use libp2p::floodsub::Event;
8use libp2p::swarm::NetworkBehaviour;
9use std::fmt::Debug;
10
11impl<X, C: NetworkBehaviour, S, T, K> ConnexaTask<X, C, S, T, K>
12where
13    X: Default + Send + 'static,
14    C: Send,
15    C::ToSwarm: Debug,
16    S: Store,
17{
18    pub fn process_floodsub_command(&mut self, command: FloodsubCommand) {
19        let swarm = self.swarm.as_mut().unwrap();
20
21        match command {
22            FloodsubCommand::Subscribe { topic, resp, .. } => {
23                let Some(pubsub) = swarm.behaviour_mut().floodsub.as_mut() else {
24                    let _ = resp.send(Err(Error::Disabled {
25                        protocol: Protocol::Floodsub,
26                    }));
27                    return;
28                };
29
30                match pubsub.subscribe(topic.clone()) {
31                    true => {
32                        let _ = resp.send(Ok(()));
33                    }
34                    false => {
35                        let _ = resp.send(Err(Error::Floodsub(
36                            floodsub::Error::AlreadySubscribed(topic),
37                        )));
38                    }
39                }
40            }
41            FloodsubCommand::Unsubscribe { topic, resp, .. } => {
42                let Some(pubsub) = swarm.behaviour_mut().floodsub.as_mut() else {
43                    let _ = resp.send(Err(Error::Disabled {
44                        protocol: Protocol::Floodsub,
45                    }));
46                    return;
47                };
48
49                match pubsub.unsubscribe(topic.clone()) {
50                    true => {
51                        let _ = resp.send(Ok(()));
52                    }
53                    false => {
54                        let _ =
55                            resp.send(Err(Error::Floodsub(floodsub::Error::NotSubscribed(topic))));
56                    }
57                }
58            }
59            FloodsubCommand::Publish(pubsub_type, resp) => {
60                let Some(pubsub) = swarm.behaviour_mut().floodsub.as_mut() else {
61                    let _ = resp.send(Err(Error::Disabled {
62                        protocol: Protocol::Floodsub,
63                    }));
64                    return;
65                };
66
67                match pubsub_type {
68                    PubsubFloodsubPublish::Publish { topic, data } => {
69                        pubsub.publish(topic, data);
70                    }
71                    PubsubFloodsubPublish::PublishAny { topic, data } => {
72                        pubsub.publish_any(topic, data);
73                    }
74                    PubsubFloodsubPublish::PublishMany { topics, data } => {
75                        pubsub.publish_many(topics, data);
76                    }
77                    PubsubFloodsubPublish::PublishManyAny { topics, data } => {
78                        pubsub.publish_many_any(topics, data);
79                    }
80                }
81
82                let _ = resp.send(Ok(()));
83            }
84            FloodsubCommand::FloodsubListener { topic, resp } => {
85                if !swarm.behaviour_mut().floodsub.is_enabled() {
86                    let _ = resp.send(Err(Error::Disabled {
87                        protocol: Protocol::Floodsub,
88                    }));
89                    return;
90                }
91
92                let topic = libp2p::floodsub::Topic::new(topic);
93
94                let (tx, rx) = mpsc::channel(50);
95
96                self.floodsub_listener.entry(topic).or_default().push(tx);
97
98                let _ = resp.send(Ok(rx));
99            }
100            FloodsubCommand::AddNodeToPartialView { peer_id, resp } => {
101                let Some(pubsub) = swarm.behaviour_mut().floodsub.as_mut() else {
102                    let _ = resp.send(Err(Error::Disabled {
103                        protocol: Protocol::Floodsub,
104                    }));
105                    return;
106                };
107
108                pubsub.add_node_to_partial_view(peer_id);
109                let _ = resp.send(Ok(()));
110            }
111            FloodsubCommand::RemoveNodeFromPartialView { peer_id, resp } => {
112                let Some(pubsub) = swarm.behaviour_mut().floodsub.as_mut() else {
113                    let _ = resp.send(Err(Error::Disabled {
114                        protocol: Protocol::Floodsub,
115                    }));
116                    return;
117                };
118
119                pubsub.remove_node_from_partial_view(&peer_id);
120                let _ = resp.send(Ok(()));
121            }
122        }
123    }
124
125    pub fn process_floodsub_event(&mut self, event: Event) {
126        let (topics, event) = match event {
127            Event::Message(libp2p::floodsub::FloodsubMessage {
128                source,
129                data,
130                sequence_number,
131                topics,
132            }) => {
133                let message = FloodsubMessage {
134                    source,
135                    data,
136                    sequence_number,
137                };
138
139                let event = FloodsubEvent::Message { message };
140
141                (topics, event)
142            }
143            Event::Subscribed { peer_id, topic } => {
144                let event = FloodsubEvent::Subscribed { peer_id };
145                (vec![topic], event)
146            }
147            Event::Unsubscribed { peer_id, topic } => {
148                let event = FloodsubEvent::Unsubscribed { peer_id };
149                (vec![topic], event)
150            }
151        };
152
153        for topic in topics {
154            let Some(chs) = self.floodsub_listener.get_mut(&topic) else {
155                continue;
156            };
157
158            // Should we retain the list too for any closed channels?
159
160            for ch in chs {
161                let _ = ch.try_send(event.clone());
162            }
163        }
164    }
165}
connexa/task/floodsub.rs

connexa/task/
floodsub.rs
