rs_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.
49pub struct Dispatcher {
50	channels: Arc<SenderChannels>,
51}
52
53impl Dispatcher {
54	/// Creates a new dispatcher
55	pub fn new(channels: Arc<SenderChannels>) -> Self {
56		Dispatcher {
57			channels: channels,
58		}
59	}
60
61	/// Pushes message to given peer
62	pub fn push_message(&self, peer_addr: &SocketAddr, msg: String) -> Result<(), PushMessageError> {
63		let mut channels = self.channels.lock();
64
65		match channels.get_mut(peer_addr) {
66			Some(channel) => {
67				// todo: maybe async here later?
68				try!(channel.send(msg).wait().map_err(|e| PushMessageError::from(e)));
69				Ok(())
70			},
71			None => {
72				return Err(PushMessageError::NoSuchPeer);
73			}
74		}
75	}
76
77	/// Returns `true` if the peer is still connnected
78	pub fn is_connected(&self, socket_addr: &SocketAddr) -> bool {
79		self.channels.lock().contains_key(socket_addr)
80	}
81
82	/// Returns current peer count.
83	pub fn peer_count(&self) -> usize {
84		self.channels.lock().len()
85	}
86}
87
88impl<S: Stream<Item=String, Error=std::io::Error>> Stream for PeerMessageQueue<S> {
89
90	type Item = String;
91	type Error = std::io::Error;
92
93	fn poll(&mut self) -> Poll<Option<String>, std::io::Error> {
94		// check if we have response pending
95		match self.up.poll() {
96			Ok(Async::Ready(Some(val))) => {
97				return Ok(Async::Ready(Some(val)));
98			},
99			Ok(Async::Ready(None)) => {
100				// this will ensure that this polling will end when incoming i/o stream ends
101				return Ok(Async::Ready(None));
102			},
103			_ => {}
104		}
105
106		match self.receiver.poll() {
107			Ok(result) => Ok(result),
108			Err(send_err) => {
109				// not sure if it can ever happen
110				warn!("MPSC send error: {:?}", send_err);
111				Err(std::io::Error::from(std::io::ErrorKind::Other))
112			}
113		}
114	}
115}
rs_jsonrpc_tcp_server/dispatch.rs

rs_jsonrpc_tcp_server/
dispatch.rs
