gossip: futures 03 Receiver

core/finality-grandpa/src/communication/mod.rs

@@ -31,6 +31,7 @@ use std::sync::Arc;
 
 use futures::prelude::*;
 use futures::sync::{oneshot, mpsc};
+use futures03::stream::{StreamExt, TryStreamExt};
 use grandpa::Message::{Prevote, Precommit, PrimaryPropose};
 use grandpa::{voter, voter_set::VoterSet};
 use log::{debug, trace};
@@ -145,12 +146,19 @@ impl<B, S, H> Network<B> for Arc<NetworkService<B, S, H>> where
 	S: network::specialization::NetworkSpecialization<B>,
 	H: network::ExHashT,
 {
-	type In = NetworkStream;
+	type In = NetworkStream<
+		Box<dyn Stream<Item = network_gossip::TopicNotification, Error = ()> + Send + 'static>,
+	>;
 
+	#[allow(deprecated)]
 	fn messages_for(&self, topic: B::Hash) -> Self::In {
 		let (tx, rx) = oneshot::channel();
 		self.with_gossip(move |gossip, _| {
-			let inner_rx = gossip.messages_for(GRANDPA_ENGINE_ID, topic);
+			let inner_rx = gossip
+				.messages_for(GRANDPA_ENGINE_ID, topic)
+				.map(|x| Ok(x))
+				.compat()
+				.boxed(); // Box::new() will break
 			let _ = tx.send(inner_rx);
 		});
 		NetworkStream { outer: rx, inner: None }
@@ -203,12 +211,15 @@ impl<B, S, H> Network<B> for Arc<NetworkService<B, S, H>> where
 }
 
 /// A stream used by NetworkBridge in its implementation of Network.
-pub struct NetworkStream {
-	inner: Option<mpsc::UnboundedReceiver<network_gossip::TopicNotification>>,
-	outer: oneshot::Receiver<mpsc::UnboundedReceiver<network_gossip::TopicNotification>>
+pub struct NetworkStream<R> {
+	inner: Option<R>,
+	outer: oneshot::Receiver<R>,
 }
 
-impl Stream for NetworkStream {
+impl<R> Stream for NetworkStream<R>
+where
+	R: Stream<Item = network_gossip::TopicNotification, Error = ()>,
+{
 	type Item = network_gossip::TopicNotification;
 	type Error = ();
 
@@ -221,9 +232,9 @@ impl Stream for NetworkStream {
 				let poll_result = inner.poll();
 				self.inner = Some(inner);
 				poll_result
-			},
+			}
 			Ok(futures::Async::NotReady) => Ok(futures::Async::NotReady),
-			Err(_) => Err(())
+			Err(_) => Err(()),
 		}
 	}
 }

core/network/src/protocol/consensus_gossip.rs

@@ -48,7 +48,7 @@ use std::sync::Arc;
 use std::iter;
 use std::time;
 use log::{trace, debug};
-use futures::sync::mpsc;
+use futures03::channel::mpsc;
 use lru_cache::LruCache;
 use libp2p::PeerId;
 use sr_primitives::traits::{Block as BlockT, Hash, HashFor};
@@ -608,7 +608,9 @@ impl<B: BlockT> Validator<B> for DiscardAll {
 #[cfg(test)]
 mod tests {
 	use sr_primitives::testing::{H256, Block as RawBlock, ExtrinsicWrapper};
-	use futures::Stream;
+	use futures03::stream::StreamExt;
+	use futures03::future::{FutureExt, TryFutureExt};
+	use tokio::runtime::current_thread::Runtime;
 
 	use super::*;
 
@@ -627,6 +629,13 @@ mod tests {
 		}
 	}
 
+	macro_rules! stream_assert_eq {
+		($stream: expr, $expected_value: expr) => {
+			let mut rt = Runtime::new().unwrap(); // use tokio 0.1's runtime on stable rust
+			assert_eq!(rt.block_on($stream.map(|x| -> Result<_, ()> { Ok(x) }).compat()).unwrap(), $expected_value);
+		};
+	}
+
 	struct AllowAll;
 	impl Validator<Block> for AllowAll {
 		fn validate(
@@ -692,18 +701,16 @@ mod tests {
 
 	#[test]
 	fn message_stream_include_those_sent_before_asking_for_stream() {
-		use futures::Stream;
-
 		let mut consensus = ConsensusGossip::<Block>::new();
 		consensus.register_validator_internal([0, 0, 0, 0], Arc::new(AllowAll));
 
 		let message = ConsensusMessage { data: vec![4, 5, 6], engine_id: [0, 0, 0, 0] };
 		let topic = HashFor::<Block>::hash(&[1,2,3]);
 
 		consensus.register_message(topic, message.clone());
-		let stream = consensus.messages_for([0, 0, 0, 0], topic);
+		let mut stream = consensus.messages_for([0, 0, 0, 0], topic);
 
-		assert_eq!(stream.wait().next(), Some(Ok(TopicNotification { message: message.data, sender: None })));
+		stream_assert_eq!(stream.next(), Some(TopicNotification { message: message.data, sender: None }));
 	}
 
 	#[test]
@@ -730,11 +737,11 @@ mod tests {
 
 		consensus.register_message(topic, message.clone());
 
-		let stream1 = consensus.messages_for([0, 0, 0, 0], topic);
-		let stream2 = consensus.messages_for([0, 0, 0, 0], topic);
+		let mut stream1 = consensus.messages_for([0, 0, 0, 0], topic);
+		let mut stream2 = consensus.messages_for([0, 0, 0, 0], topic);
 
-		assert_eq!(stream1.wait().next(), Some(Ok(TopicNotification { message: message.data.clone(), sender: None })));
-		assert_eq!(stream2.wait().next(), Some(Ok(TopicNotification { message: message.data, sender: None })));
+		stream_assert_eq!(stream1.next(), Some(TopicNotification { message: message.data.clone(), sender: None }));
+		stream_assert_eq!(stream2.next(), Some(TopicNotification { message: message.data, sender: None }));
 	}
 
 	#[test]
@@ -749,10 +756,10 @@ mod tests {
 		consensus.register_message(topic, msg_a);
 		consensus.register_message(topic, msg_b);
 
-		let mut stream = consensus.messages_for([0, 0, 0, 0], topic).wait();
+		let mut stream = consensus.messages_for([0, 0, 0, 0], topic);
 
-		assert_eq!(stream.next(), Some(Ok(TopicNotification { message: vec![1, 2, 3], sender: None })));
+		stream_assert_eq!(stream.next(), Some(TopicNotification { message: vec![1, 2, 3], sender: None }));
 		let _ = consensus.live_message_sinks.remove(&([0, 0, 0, 0], topic));
-		assert_eq!(stream.next(), None);
+		stream_assert_eq!(stream.next(), None);
 	}
 }