update tokio-stream to 0.1.3 and use BroadcastStream (sigp#2212)

## Issue Addressed Resolves sigp#2189 ## Proposed Changes use tokio's `BroadcastStream` ## Additional Info N/A Co-authored-by: realbigsean <[email protected]>
paulhauner · Mar 1, 2021 · ed9b245 · ed9b245
1 parent baef1db
commit ed9b245
Show file tree

Hide file tree

Showing 5 changed files with 20 additions and 87 deletions.
diff --git a/Cargo.lock b/Cargo.lock
diff --git a/beacon_node/http_api/Cargo.toml b/beacon_node/http_api/Cargo.toml
@@ -8,7 +8,7 @@ edition = "2018"
 warp = { git = "https://github.com/paulhauner/warp ", branch = "cors-wildcard" }
 serde = { version = "1.0.116", features = ["derive"] }
 tokio = { version = "1.1.0", features = ["macros","sync"] }
-tokio-stream = "0.1.2"
+tokio-stream = { version = "0.1.3", features = ["sync"] }
 tokio-util = "0.6.3"
 parking_lot = "0.11.0"
 types = { path = "../../consensus/types" }

diff --git a/beacon_node/http_api/src/broadcast_stream.rs b/beacon_node/http_api/src/broadcast_stream.rs
diff --git a/beacon_node/http_api/src/lib.rs b/beacon_node/http_api/src/lib.rs
@@ -7,7 +7,6 @@
 
 mod beacon_proposer_cache;
 mod block_id;
-mod broadcast_stream;
 mod metrics;
 mod state_id;
 mod validator_inclusion;
@@ -36,7 +35,7 @@ use std::future::Future;
 use std::net::{Ipv4Addr, SocketAddr, SocketAddrV4};
 use std::sync::Arc;
 use tokio::sync::mpsc::UnboundedSender;
-use tokio_stream::StreamExt;
+use tokio_stream::{wrappers::BroadcastStream, StreamExt};
 use types::{
     Attestation, AttestationDuty, AttesterSlashing, CloneConfig, CommitteeCache, Epoch, EthSpec,
     Hash256, ProposerSlashing, PublicKey, PublicKeyBytes, RelativeEpoch, SignedAggregateAndProof,
@@ -2405,23 +2404,22 @@ pub fn serve<T: BeaconChainTypes>(
                                 }
                             };
 
-                            receivers.push(broadcast_stream::BroadcastStream::new(receiver).map(
-                                |msg| {
-                                    match msg {
-                                        Ok(data) => Event::default()
-                                            .event(data.topic_name())
-                                            .json_data(data)
-                                            .map_err(|e| {
-                                                warp_utils::reject::server_sent_event_error(
-                                                    format!("{:?}", e),
-                                                )
-                                            }),
-                                        Err(e) => Err(warp_utils::reject::server_sent_event_error(
-                                            format!("{:?}", e),
-                                        )),
-                                    }
-                                },
-                            ));
+                            receivers.push(BroadcastStream::new(receiver).map(|msg| {
+                                match msg {
+                                    Ok(data) => Event::default()
+                                        .event(data.topic_name())
+                                        .json_data(data)
+                                        .map_err(|e| {
+                                            warp_utils::reject::server_sent_event_error(format!(
+                                                "{:?}",
+                                                e
+                                            ))
+                                        }),
+                                    Err(e) => Err(warp_utils::reject::server_sent_event_error(
+                                        format!("{:?}", e),
+                                    )),
+                                }
+                            }));
                         }
                     } else {
                         return Err(warp_utils::reject::custom_server_error(

diff --git a/slasher/service/Cargo.toml b/slasher/service/Cargo.toml
@@ -15,5 +15,5 @@ slot_clock = { path = "../../common/slot_clock" }
 state_processing = { path = "../../consensus/state_processing" }
 task_executor = { path = "../../common/task_executor" }
 tokio = { version = "1.1.0", features = ["full"] }
-tokio-stream = "0.1.2"
+tokio-stream = "0.1.3"
 types = { path = "../../consensus/types" }