pezkuwichain/pezkuwi-telemetry
1
0
Fork 0
mirror of https://github.com/pezkuwichain/pezkuwi-telemetry.git synced 2026-05-30 01:51:08 +00:00
Code Issues Packages Projects Releases Wiki Activity

simplify feed sending a little

Browse Source
This commit is contained in:
James Wilson
2021-06-25 17:39:22 +01:00
parent 89dfad5bbe
commit 00c6e4f4c5
1 changed files with 23 additions and 29 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/telemetry/src/aggregator/inner_loop.rs
+23 -29
View File
@@ -197,17 +197,12 @@ impl InnerLoop {
&loc.city &loc.city
)); ));
if let Some(bytes) = feed_message_serializer.into_finalized() { let chain_label = self.node_state
let chain_label = self.node_state .get_node_chain(node_id)
.get_node_chain(node_id) .map(|chain| chain.label().to_owned());
.map(|chain| chain.label());
if let Some(chain_label) = chain_label { if let Some(chain_label) = chain_label {
// Don't hold onto lifetime from self because we call a mut fn next: self.finalize_and_broadcast_to_chain_feeds(&chain_label, feed_message_serializer).await;
let label = chain_label.to_owned();
// Update location for any feeds subscribed to the node's chain.
self.broadcast_to_chain_feeds(&label, ToFeedWebsocket::Bytes(bytes)).await;
}
} }
} }
} }
@@ -251,13 +246,7 @@ impl InnerLoop {
// Tell chain subscribers about the node we've just added: // Tell chain subscribers about the node we've just added:
let mut feed_messages_for_chain = FeedMessageSerializer::new(); let mut feed_messages_for_chain = FeedMessageSerializer::new();
feed_messages_for_chain.push(feed_message::AddedNode(node_id, &details.node)); feed_messages_for_chain.push(feed_message::AddedNode(node_id, &details.node));
self.finalize_and_broadcast_to_chain_feeds(&old_chain_label, feed_messages_for_chain).await;
if let Some(bytes) = feed_messages_for_chain.into_finalized() {
self.broadcast_to_chain_feeds(
&new_chain_label,
ToFeedWebsocket::Bytes(bytes)
).await;
}
// Tell everybody about the new node count and potential rename: // Tell everybody about the new node count and potential rename:
let mut feed_messages_for_all = FeedMessageSerializer::new(); let mut feed_messages_for_all = FeedMessageSerializer::new();
@@ -265,11 +254,7 @@ impl InnerLoop {
feed_messages_for_all.push(feed_message::RemovedChain(&old_chain_label)); feed_messages_for_all.push(feed_message::RemovedChain(&old_chain_label));
} }
feed_messages_for_all.push(feed_message::AddedChain(&new_chain_label, chain_node_count)); feed_messages_for_all.push(feed_message::AddedChain(&new_chain_label, chain_node_count));
self.finalize_and_broadcast_to_all_feeds(feed_messages_for_all).await;
if let Some(bytes) = feed_messages_for_all.into_finalized() {
let msg = ToFeedWebsocket::Bytes(bytes);
self.broadcast_to_all_feeds(msg).await;
}
// Ask for the grographical location of the node. // Ask for the grographical location of the node.
// Currently we only geographically locate IPV4 addresses so ignore IPV6. // Currently we only geographically locate IPV4 addresses so ignore IPV6.
@@ -486,14 +471,9 @@ impl InnerLoop {
&mut feed_messages_for_all &mut feed_messages_for_all
); );
} }
if let Some(bytes) = feed_messages_for_chain.into_finalized() { self.finalize_and_broadcast_to_chain_feeds(&chain_label, feed_messages_for_chain).await;
self.broadcast_to_chain_feeds(&chain_label, ToFeedWebsocket::Bytes(bytes)).await;
}
}
if let Some(bytes) = feed_messages_for_all.into_finalized() {
self.broadcast_to_all_feeds(ToFeedWebsocket::Bytes(bytes)).await;
} }
self.finalize_and_broadcast_to_all_feeds(feed_messages_for_all).await;
} }
/// Remove a single node by its ID, pushing any messages we'd want to send /// Remove a single node by its ID, pushing any messages we'd want to send
@@ -538,6 +518,13 @@ impl InnerLoop {
} }
} }
/// Finalize a [`FeedMessageSerializer`] and broadcast the result to feeds for the chain.
async fn finalize_and_broadcast_to_chain_feeds(&mut self, chain: &str, serializer: FeedMessageSerializer) {
if let Some(bytes) = serializer.into_finalized() {
self.broadcast_to_chain_feeds(chain, ToFeedWebsocket::Bytes(bytes)).await;
}
}
/// Send a message to all chain feeds. /// Send a message to all chain feeds.
async fn broadcast_to_chain_feeds(&mut self, chain: &str, message: ToFeedWebsocket) { async fn broadcast_to_chain_feeds(&mut self, chain: &str, message: ToFeedWebsocket) {
if let Some(feeds) = self.chain_to_feed_conn_ids.get(chain) { if let Some(feeds) = self.chain_to_feed_conn_ids.get(chain) {
@@ -551,6 +538,13 @@ impl InnerLoop {
} }
} }
/// Finalize a [`FeedMessageSerializer`] and broadcast the result to all feeds
async fn finalize_and_broadcast_to_all_feeds(&mut self, serializer: FeedMessageSerializer) {
if let Some(bytes) = serializer.into_finalized() {
self.broadcast_to_all_feeds(ToFeedWebsocket::Bytes(bytes)).await;
}
}
/// Send a message to everybody. /// Send a message to everybody.
async fn broadcast_to_all_feeds(&mut self, message: ToFeedWebsocket) { async fn broadcast_to_all_feeds(&mut self, message: ToFeedWebsocket) {
for chan in self.feed_channels.values_mut() { for chan in self.feed_channels.values_mut() {