Drop timeout if max concurrent timeouts reached (paritytech#324)

* closes paritytech#322

* refactored error handling

* proper loop handling

* schedule SuspendableStream on timer error

Author: seunlanlege

server-utils/src/suspendable_stream.rs

@@ -40,30 +40,34 @@ impl<S, I> Stream for SuspendableStream<S>
 	type Error = ();
 
 	fn poll(&mut self) -> Result<Async<Option<Self::Item>>, ()> {
-		if let Some(mut timeout) = self.timeout.take() {
-			match timeout.poll() {
-				Ok(Async::Ready(_)) => {}
-				Ok(Async::NotReady) => {
-					self.timeout = Some(timeout);
-					return Ok(Async::NotReady);
+		loop {
+			if let Some(mut timeout) = self.timeout.take() {
+				match timeout.poll() {
+					Ok(Async::Ready(_)) => {}
+					Ok(Async::NotReady) => {
+						self.timeout = Some(timeout);
+						return Ok(Async::NotReady);
+					}
+					Err(err) => {
+						warn!("Timeout error {:?}", err);
+						task::current().notify();
+						return Ok(Async::NotReady);
+					}
 				}
-				Err(_) => unreachable!("Polling a delay shouldn't yield any errors; qed")
 			}
-		}
 
-		loop {
 			match self.stream.poll() {
 				Ok(item) => {
 					if self.next_delay > self.initial_delay {
 						self.next_delay = self.initial_delay;
 					}
 					return Ok(item)
-				},
-				Err(ref e) => if connection_error(e) {
-					warn!("Connection Error: {:?}", e);
-					continue
 				}
-				Err(err) => {
+				Err(ref err) => {
+					if connection_error(err) {
+						warn!("Connection Error: {:?}", err);
+						continue
+					}
 					self.next_delay = if self.next_delay < self.max_delay {
 						self.next_delay * 2
 					} else {