[Re-]Implement Automatic Reconnection

[aatxe]

Automatic reconnection is clearly a desirable feature, and implementing it correctly should be a lot easier with the new tokio backend in 0.12.0.

[DoumanAsh]

Just to clarify: is current implementation is not able to reconnect or is it lacking something?

[aatxe]

Prior to 0.12.0, there was a broken implementation. As of 0.12.0, there is no automatic reconnection whatsoever. It should be possible for a library consumer to implement it themselves though in a fairly straightforward way. If you have set up your own event loop with tokio-core (there's an example of this here), you should be able to wrap this with error-handling that just goes through the process again. It wouldn't do any state restoration beyond what is specified in the configuration though. For example, you wouldn't rejoin channels that you joined during the course of execution.

From my perspective, the perfect automatic reconnection implementation for this library would restore all possible channel membership (using the chanlists feature). It also wouldn't require the exposure of tokio internals as above. Reconnecting to just what's specified in the config might be a nice stepping stone though. It's been suggested to me a few times that this could all be done inside the transport (i.e. src/client/transport.rs), but I have not yet grokked how to do it.

[DoumanAsh]

I'm planning on using your library to implement IRC bot so I wanted to have clarifications. Thanks for it and examples.

[aatxe]

To anyone looking here, this is less of an issue today than in the past (perhaps to the point where we don't need it in the library, but I'm not committing one way or the other yet).

As of 0.13, the following template should handle reconnection well and is only slightly more code than the basic template (although, adding exponential backoff or something is likely a good idea):

extern crate irc;

use irc::client::prelude::*;

fn main() {
    // We can also load the Config at runtime via Config::load("path/to/config.toml")
    let config = Config {
        nickname: Some("the-irc-crate".to_owned()),
        server: Some("irc.pdgn.co".to_owned()),
        channels: Some(vec!["#test".to_owned()]),
        ..Config::default()
    };

    while let Err(e) = implementation(&config) {
        eprintln!("{}", e);
        // One might wish to add a backoff strategy here to prevent reconnection attempts in a tight loop.
    }
}

fn implementation(config: &Config) -> irc::error::Result<()> {
    let mut reactor = IrcReactor::new()?;
    let client = reactor.prepare_client_and_connect(config)?;
    client.identify()?;

    reactor.register_client_with_handler(client, |client, message| {
        print!("{}", message);
        // And here we can do whatever we want with the messages.
        Ok(())
    });

    reactor.run()
}

[cemeyer]

As of 0.14, the 0.13 example no longer works; everything runs in the implicit Tokio global reactor and these other symbols are no longer exposed. I have a complete hack to automatically reconnect and provide a consistent stream. There is some difficulty in managing the lifetime of the client, when another task may want to send commands to it (like: send_quit()). I'm sure this unsafe construction isn't the best way to do it and it seems likely the best place to handle it would be inside the client library itself.

static mut irc_client: Option<Box<Client>> = None;

#[tokio::main]
async fn main() -> Result<(), failure::Error> {
    ...

    let (tx, irc_stream) = mpsc::unbounded_channel();
    let irc_tasklet = {
        let irc_client_h = unsafe { &mut irc_client };
        tokio::task::spawn(async move {
            loop {
                // Sleep-loop until creating an IRC client succeeds.  I think it's async, so I have
                // no idea how it can fail.  But go ahead and gracefully retry.
                let mut new_client = {
                    let mut r;
                    loop {
                        r = create_irc(&irc_user, &irc_nickserv_pass).await;
                        if r.is_ok() {
                            break;
                        }
                        log::error!("Failed to create irc client, sleeping 60 seconds: {}", r.unwrap_err());
                        tokio::time::delay_for(std::time::Duration::from_secs(60)).await;
                    }
                    r.unwrap()
                };
                *irc_client_h = Some(new_client);
                let irc_stream = irc_client_h.as_mut().unwrap().as_mut().stream().unwrap();

                // Fuse IRC stream /after/ first fatal error.
                let mut conn_error = false;
                let irc_stream = irc_stream.take_while(move |x| {
                    if conn_error {
                        return futures::future::ready(false);
                    }
                    if x.is_err() {
                        conn_err = match x.as_ref().unwrap_err() {
                            irc::error::Error::PingTimeout => true,
                            irc::error::Error::Io(_) => true,
                            irc::error::Error::Tls(_) => true,
                            _ => false
                        };
                    }
                    return futures::future::ready(true);
                });

                // Shovel IRC events from the current client's stream into our queue, until a fatal
                // error was detected.
                irc_stream.for_each(|e| {
                    tx.send(e).unwrap();
                    futures::future::ready(())
                }).await;

                log::error!("IRC fatal error, trying to reconnect.");

                *irc_client_h = None;
                // Implicit drop of irc_stream, irc_client.
            }
        })
    };

    ...

    unsafe {
        if irc_client.is_some() {
            irc_client.as_ref().unwrap().send_quit("quitting")?;
        }
    }

    ...

    irc_tasklet.await?;

    ...