gnosis/lighthouse
1
0
Fork 0
mirror of https://github.com/sigp/lighthouse.git synced 2026-03-03 00:31:50 +00:00
Code Issues Packages Projects Releases Wiki Activity

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

Browse Source 
## Issue Addressed

Resolves #2189 

## Proposed Changes

use tokio's `BroadcastStream`

## Additional Info

N/A


Co-authored-by: realbigsean <seananderson33@gmail.com>
This commit is contained in:
realbigsean
2021-03-01 01:58:05 +00:00
parent baef1db40f
commit ed9b245de0
5 changed files with 20 additions and 87 deletions
Download Patch File Download Diff File Expand all files Collapse all files

36
beacon_node/http_api/src/lib.rs
View File

@@ -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(