--autoreconnect-delay-millis

Addresses #66

src/lib.rs

@@ -21,6 +21,7 @@ extern crate tokio_reactor;
 extern crate tokio_tcp;
 extern crate tokio_udp;
 extern crate tokio_codec;
+extern crate tokio_timer;
 extern crate websocket;
 extern crate websocket_base;
 extern crate http_bytes;
@@ -41,8 +42,6 @@ extern crate smart_default;
 #[macro_use]
 extern crate derivative;
 
-extern crate tokio_timer;
-
 use futures::future::Future;
 use tokio_io::{AsyncRead, AsyncWrite};
 

src/main.rs

@@ -447,6 +447,10 @@ struct Opt {
  /// Maximum number of messages to copy in the other direction.
  #[structopt(long = "max-messages-rev")]
  max_messages_rev: Option<usize>,
+
+ /// [A] Delay before reconnect attempt for `autoreconnect:` overlay.
+ #[structopt(long = "--autoreconnect-delay-millis", default_value="20")]
+ autoreconnect_delay_millis: u64,
 }
 
 // TODO: make it byte-oriented/OsStr?
@@ -688,6 +692,7 @@ fn run() -> Result<()> {
  no_exit_on_zeromsg
  max_messages
  max_messages_rev
+ autoreconnect_delay_millis
  );
  #[cfg(feature = "ssl")]
  {

src/options.rs

@@ -81,4 +81,6 @@ pub struct Options {
  pub request_uri: Option<http::Uri>,
  pub request_method: Option<http::Method>,
  pub request_headers: Vec<(http::header::HeaderName, http::header::HeaderValue)>,
+
+ pub autoreconnect_delay_millis: u64,
 }

src/reconnect_peer.rs

@@ -59,6 +59,8 @@ struct State {
  n: Option<BoxedNewPeerFuture>,
  cp: ConstructParams,
  aux: State2,
+ reconnect_delay: std::time::Duration,
+ ratelimiter: Option<tokio_timer::Delay>,
 }
 
 /// This implementation's poll is to be reused many times, both after returning item and error
@@ -73,6 +75,16 @@ impl State {
  let aux = &mut self.aux;
 
  loop {
+ if let Some(delay) = self.ratelimiter.as_mut() {
+ match delay.poll() {
+ Ok(Async::Ready(_)) => {
+ debug!("Waited for reconnect");
+ self.ratelimiter = None;
+ }
+ Err(e) => error!("tokio-timer's Delay: {}", e),
+ Ok(Async::NotReady) => return Ok(Async::NotReady),
+ }
+ }
  let cp = self.cp.clone();
  if let Some(ref mut p) = *pp {
  return Ok(Async::Ready(p));
@@ -98,8 +110,13 @@ impl State {
 
  if !aux.already_warned {
  aux.already_warned = true;
- error!("Reconnecting failed. Trying again in tight endless loop.");
+ warn!("Reconnecting failed. Further failed reconnects announncements will have lower severity.");
+ } else {
+ info!("Reconnecting failed.");
  }
+
+ self.ratelimiter = Some(tokio_timer::Delay::new(std::time::Instant::now() + self.reconnect_delay));
+ continue;
  }
  }
  }
@@ -203,12 +220,15 @@ impl AsyncWrite for PeerHandle {
 }
 
 pub fn autoreconnector(s: Rc<dyn Specifier>, cp: ConstructParams) -> BoxedNewPeerFuture {
+ let reconnect_delay = std::time::Duration::from_millis(cp.program_options.autoreconnect_delay_millis);
  let s = Rc::new(RefCell::new(State {
  cp,
  s,
  p: None,
  n: None,
  aux: Default::default(),
+ reconnect_delay,
+ ratelimiter: None,
  }));
  let ph1 = PeerHandle(s.clone());
  let ph2 = PeerHandle(s);