bitconch_jsonrpc_tcp_server/
dispatch.rs

1use std;
2use std::collections::HashMap;
3use std::net::SocketAddr;
4use std::sync::{Arc};
5
6use jsonrpc::futures::{Stream, Poll, Async, Sink, Future};
7use jsonrpc::futures::sync::mpsc;
8
9use parking_lot::Mutex;
10
11pub type SenderChannels = Mutex<HashMap<SocketAddr, mpsc::Sender<String>>>;
12
13pub struct PeerMessageQueue<S: Stream> {
14	up: S,
15	receiver: mpsc::Receiver<String>,
16	_addr: SocketAddr,
17}
18
19impl<S: Stream> PeerMessageQueue<S> {
20	pub fn new(
21		response_stream: S,
22		receiver: mpsc::Receiver<String>,
23		addr: SocketAddr,
24	) -> Self {
25		PeerMessageQueue {
26			up: response_stream,
27			receiver: receiver,
28			_addr: addr,
29		}
30	}
31}
32
33/// Push Message Error
34#[derive(Debug)]
35pub enum PushMessageError {
36	/// Invalid peer
37	NoSuchPeer,
38	/// Send error
39	Send(mpsc::SendError<String>)
40}
41
42impl From<mpsc::SendError<String>> for PushMessageError {
43	fn from(send_err: mpsc::SendError<String>) -> Self {
44		PushMessageError::Send(send_err)
45	}
46}
47
48/// Peer-messages dispatcher.
49#[derive(Clone)]
50pub struct Dispatcher {
51	channels: Arc<SenderChannels>,
52}
53
54impl Dispatcher {
55	/// Creates a new dispatcher
56	pub fn new(channels: Arc<SenderChannels>) -> Self {
57		Dispatcher {
58			channels: channels,
59		}
60	}
61
62	/// Pushes message to given peer
63	pub fn push_message(&self, peer_addr: &SocketAddr, msg: String) -> Result<(), PushMessageError> {
64		let mut channels = self.channels.lock();
65
66		match channels.get_mut(peer_addr) {
67			Some(channel) => {
68				// todo: maybe async here later?
69				try!(channel.send(msg).wait().map_err(|e| PushMessageError::from(e)));
70				Ok(())
71			},
72			None => {
73				return Err(PushMessageError::NoSuchPeer);
74			}
75		}
76	}
77
78	/// Returns `true` if the peer is still connnected
79	pub fn is_connected(&self, socket_addr: &SocketAddr) -> bool {
80		self.channels.lock().contains_key(socket_addr)
81	}
82
83	/// Returns current peer count.
84	pub fn peer_count(&self) -> usize {
85		self.channels.lock().len()
86	}
87}
88
89impl<S: Stream<Item=String, Error=std::io::Error>> Stream for PeerMessageQueue<S> {
90
91	type Item = String;
92	type Error = std::io::Error;
93
94	fn poll(&mut self) -> Poll<Option<String>, std::io::Error> {
95		// check if we have response pending
96		match self.up.poll() {
97			Ok(Async::Ready(Some(val))) => {
98				return Ok(Async::Ready(Some(val)));
99			},
100			Ok(Async::Ready(None)) => {
101				// this will ensure that this polling will end when incoming i/o stream ends
102				return Ok(Async::Ready(None));
103			},
104			_ => {}
105		}
106
107		match self.receiver.poll() {
108			Ok(result) => Ok(result),
109			Err(send_err) => {
110				// not sure if it can ever happen
111				warn!("MPSC send error: {:?}", send_err);
112				Err(std::io::Error::from(std::io::ErrorKind::Other))
113			}
114		}
115	}
116}
bitconch_jsonrpc_tcp_server/dispatch.rs

bitconch_jsonrpc_tcp_server/
dispatch.rs
