// Copyright 2015 MaidSafe.net limited.
//
// This SAFE Network Software is licensed to you under (1) the MaidSafe.net Commercial License,
// version 1.0 or later, or (2) The General Public License (GPL), version 3, depending on which
// licence you accepted on initial access to the Software (the "Licences").
//
// By contributing code to the SAFE Network Software, or to this project generally, you agree to be
// bound by the terms of the MaidSafe Contributor Agreement, version 1.0.  This, along with the
// Licenses can be found in the root directory of this project at LICENSE, COPYING and CONTRIBUTOR.
//
// Unless required by applicable law or agreed to in writing, the SAFE Network Software distributed
// under the GPL Licence is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
// KIND, either express or implied.
//
// Please review the Licences for the specific language governing permissions and limitations
// relating to use of the SAFE Network Software.

#![deny(unused)]

use std::collections::{HashMap, HashSet};
use std::hash::{Hash, Hasher, SipHasher};
use std::io;
use std::sync::{mpsc, Arc, Mutex};
use std::sync::atomic::{AtomicBool, Ordering};
use std::fmt::Write;
use service_discovery::ServiceDiscovery;
use sodiumoxide;
use sodiumoxide::crypto::box_;
use sodiumoxide::crypto::box_::{PublicKey, SecretKey};
use net2;
use nat_traversal::{MappingContext, PrivRendezvousInfo, MappedTcpSocket, PubRendezvousInfo,
                    gen_rendezvous_info, SimpleTcpHolePunchServer, tcp_punch_hole};


use sender_receiver::CrustMsg;
use connection::RaiiTcpAcceptor;
use static_contact_info::StaticContactInfo;
use config_handler::{Config, read_config_file};
use connection::{Connection, ConnectionInfo};
use error::Error;
use connection;
use bootstrap;
use bootstrap::RaiiBootstrap;
use bootstrap_handler::BootstrapHandler;
use std::time::Duration;

use event::Event;
use socket_addr::SocketAddr;
use peer_id;
use peer_id::PeerId;

/// Default beacon (service discovery) port.
pub const DEFAULT_BEACON_PORT: u16 = 5484;

pub const HEARTBEAT_PERIOD_SECS: u64 = 20;
pub const INACTIVITY_TIMEOUT_SECS: u64 = 60;

/// The result of a `Service::prepare_contact_info` call.
#[derive(Debug)]
pub struct ConnectionInfoResult {
    /// The token that was passed to `prepare_connection_info`.
    pub result_token: u32,
    /// The new contact info, if successful.
    pub result: io::Result<OurConnectionInfo>,
}

/// Contact info generated by a call to `Service::prepare_contact_info`.
#[derive(Debug)]
pub struct OurConnectionInfo {
    id: PeerId,
    tcp_info: PubRendezvousInfo,
    priv_tcp_info: PrivRendezvousInfo,
    tcp_socket: Option<net2::TcpBuilder>,
    static_contact_info: StaticContactInfo,
}

impl OurConnectionInfo {
    /// Convert our connection info to theirs so that we can give it to peer.
    pub fn to_their_connection_info(&self) -> TheirConnectionInfo {
        TheirConnectionInfo {
            tcp_info: self.tcp_info.clone(),
            static_contact_info: self.static_contact_info.clone(),
            // tcp_addrs: self.tcp_addrs.clone(),
            id: self.id,
        }
    }
}

/// Contact info used to connect to another peer.
#[derive(Debug, RustcEncodable, RustcDecodable)]
pub struct TheirConnectionInfo {
    tcp_info: PubRendezvousInfo,
    static_contact_info: StaticContactInfo,
    id: PeerId,
}

impl TheirConnectionInfo {
    /// Returns the `PeerId` of the node that created this connection info.
    pub fn id(&self) -> PeerId {
        self.id
    }
}

/// Returns a hash of the network name.
fn name_hash(network_name: &Option<String>) -> u64 {
    let mut hasher = SipHasher::new();
    debug!("Network name: {:?}", network_name);
    network_name.hash(&mut hasher);
    hasher.finish()
}

/// A structure representing a connection manager.
///
/// This abstraction has a hidden dependency on a config file. Refer to [the docs for `FileHandler`]
/// (../file_handler/struct.FileHandler.html) and [an example config file flowchart]
/// (https://github.com/maidsafe/crust/blob/master/docs/vault_config_file_flowchart.pdf) for more
/// information.
pub struct Service {
    heart_beat_timeout: Duration,
    inactivity_timeout: Duration,
    static_contact_info: Arc<Mutex<StaticContactInfo>>,
    bootstrap_cache: Arc<Mutex<BootstrapHandler>>,
    expected_peers: Arc<Mutex<HashSet<PeerId>>>,
    service_discovery: ServiceDiscovery<StaticContactInfo>,
    event_tx: ::CrustEventSender,
    bootstrap: RaiiBootstrap,
    our_keys: (PublicKey, SecretKey),
    connection_map: Arc<Mutex<HashMap<PeerId, Vec<Connection>>>>,
    mapping_context: Arc<MappingContext>,
    tcp_acceptor_port: Option<u16>,
    raii_tcp_acceptor: Option<RaiiTcpAcceptor>,
    _tcp_hole_punch_server: SimpleTcpHolePunchServer<Arc<MappingContext>>,
    is_alive: Arc<AtomicBool>,
    name_hash: u64,
}

impl Service {
    /// Constructs a service. User needs to create an asynchronous channel, and provide
    /// the sender half to this method. Receiver will receive all `Event`s from this library.
    pub fn new(event_tx: ::CrustEventSender) -> Result<Service, Error> {
        Service::with_config(event_tx, &try!(read_config_file()))
    }

    /// Constructs a service with the given config. User needs to create an asynchronous channel,
    /// and provide the sender half to this method. Receiver will receive all `Event`s from this
    /// library.
    pub fn with_config(event_tx: ::CrustEventSender, config: &Config) -> Result<Service, Error> {
        sodiumoxide::init();

        let our_keys = box_::gen_keypair();

        // Form our initial contact info
        let static_contact_info = Arc::new(Mutex::new(StaticContactInfo {
            tcp_acceptors: Vec::new(),
            tcp_mapper_servers: Vec::new(),
        }));

        let cloned_contact_info = static_contact_info.clone();
        let generator = move || unwrap_result!(cloned_contact_info.lock()).clone();
        let service_discovery =
            try!(ServiceDiscovery::new_with_generator(config.service_discovery_port
                                                            .unwrap_or(DEFAULT_BEACON_PORT),
                                                      generator));

        let mapping_context = try!(MappingContext::new()
                                       .result_log()
                                       .or_else(|e| {
                                           Err(io::Error::new(io::ErrorKind::Other,
                                                              format!("Failed to create \
                                                                       MappingContext: {}",
                                                                      e)))
                                       }));
        // Form initial peer contact infos - these will also contain echo-service addrs.
        let bootstrap_cache =
            Arc::new(Mutex::new(try!(BootstrapHandler::new(&config.bootstrap_cache_name))));
        let bootstrap_contacts = try!(bootstrap::get_known_contacts(&service_discovery,
                                                                    bootstrap_cache.clone(),
                                                                    &config));
        for peer_contact_info in bootstrap_contacts.iter() {
            mapping_context.add_simple_tcp_servers(peer_contact_info.tcp_mapper_servers.clone());
        }

        let connection_map = Arc::new(Mutex::new(HashMap::new()));

        mapping_context.add_simple_tcp_servers(config.tcp_mapper_servers.clone());
        let mapping_context = Arc::new(mapping_context);
        let name_hash = name_hash(&config.network_name);

        let heart_beat_timeout = Duration::from_secs(HEARTBEAT_PERIOD_SECS);
        let inactivity_timeout = Duration::from_secs(INACTIVITY_TIMEOUT_SECS);

        let bootstrap = RaiiBootstrap::new(bootstrap_contacts,
                                           heart_beat_timeout,
                                           inactivity_timeout,
                                           our_keys.0.clone(),
                                           event_tx.clone(),
                                           connection_map.clone(),
                                           bootstrap_cache.clone(),
                                           mapping_context.clone(),
                                           name_hash);

        let tcp_hole_punch_server = try!(SimpleTcpHolePunchServer::new(mapping_context.clone())
                                             .result_log()
                                             .or_else(|err| {
                                                 let err_str = format!("Failed to create TCP \
                                                                        hole punch server: {}",
                                                                       err);
                                                 Err(io::Error::new(io::ErrorKind::Other, err_str))
                                             }));

        {
            let mut static_contact_info = static_contact_info.lock().unwrap();
            static_contact_info.tcp_mapper_servers.extend(tcp_hole_punch_server.addresses());
        }

        let service = Service {
            heart_beat_timeout: heart_beat_timeout,
            inactivity_timeout: inactivity_timeout,
            static_contact_info: static_contact_info,
            bootstrap_cache: bootstrap_cache,
            service_discovery: service_discovery,
            expected_peers: Arc::new(Mutex::new(HashSet::new())),
            event_tx: event_tx,
            bootstrap: bootstrap,
            our_keys: our_keys,
            connection_map: connection_map,
            mapping_context: mapping_context.clone(),
            tcp_acceptor_port: config.tcp_acceptor_port,
            raii_tcp_acceptor: None,
            _tcp_hole_punch_server: tcp_hole_punch_server,
            is_alive: Arc::new(AtomicBool::new(true)),
            name_hash: name_hash,
        };

        Ok(service)
    }

    /// Stop the bootstraping procedure
    pub fn stop_bootstrap(&mut self) {
        self.bootstrap.stop();
    }

    /// Starts accepting TCP connections.
    pub fn start_listening_tcp(&mut self) -> io::Result<()> {
        // Start the TCP Acceptor
        let result = connection::start_tcp_accept(self.tcp_acceptor_port.unwrap_or(0),
                                                  self.heart_beat_timeout,
                                                  self.inactivity_timeout,
                                                  self.static_contact_info.clone(),
                                                  self.our_keys.0.clone(),
                                                  self.event_tx.clone(),
                                                  self.connection_map.clone(),
                                                  self.bootstrap_cache.clone(),
                                                  self.expected_peers.clone(),
                                                  self.mapping_context.clone(),
                                                  self.name_hash);
        self.raii_tcp_acceptor = Some(try!(result));
        Ok(())
    }

    /// Starts listening for beacon broadcasts.
    pub fn start_service_discovery(&mut self) {
        if !self.service_discovery.set_listen_for_peers(true) {
            debug!("Could not start service discovery");
        }
    }

    /// Get the hole punch servers addresses of nodes that we're connected to ordered by how likely
    /// they are to be on a seperate network.
    pub fn get_ordered_helping_nodes(&self) -> Vec<SocketAddr> {
        unimplemented!()
    }

    /// Send the given `data` to the peer with the given `PeerId`.
    ///
    /// Messages with a _lower_ value for `priority` will be sent _before_ messages with a higher
    /// number, and they are _less_ likely to be dropped if the upstream bandwidth is insufficient.
    pub fn send(&self, id: &PeerId, data: Vec<u8>, priority: u8) -> io::Result<()> {
        match unwrap_result!(self.connection_map.lock())
                  .get_mut(&id)
                  .and_then(|conns| conns.get_mut(0)) {
            None => {
                let msg = format!("No connection to peer {}", id);
                Err(io::Error::new(io::ErrorKind::Other, msg))
            }
            Some(connection) => connection.send(CrustMsg::Message(data), priority),
        }
    }

    /// Get information about our connection to a peer.
    pub fn connection_info(&self, id: &PeerId) -> Option<ConnectionInfo> {
        unwrap_result!(self.connection_map.lock())
            .get(&id)
            .and_then(|conns| conns.get(0))
            .and_then(|conn| Some(conn.get_info()))
    }

    /// Disconnect from the given peer and returns whether there was a connection at all.
    pub fn disconnect(&self, id: &PeerId) -> bool {
        debug!("dropping connection to peer {:?}", id);
        unwrap_result!(self.connection_map.lock()).remove(&id).is_some()
    }

    /// Opens a connection to a remote peer. `public_endpoint` is the endpoint
    /// of the remote peer. `udp_socket` is a socket whose public address will
    /// be used by the other peer.
    ///
    /// A rendezvous connection setup is different to the traditional BSD socket
    /// setup in which there is no client or server side. Both ends create a
    /// socket and send somehow its public address to the other peer. Once both
    /// ends know each other address, both must call this function passing the
    /// socket which possess the address used by the other peer and passing the
    /// other peer's address.
    ///
    /// On success `Event::NewPeer` with connected `PeerId` will
    /// be sent to the event channel. On failure, nothing is reported. Failed
    /// attempts are not notified back up to the caller. If the caller wants to
    /// know of a failed attempt, it must maintain a record of the attempt
    /// itself which times out if a corresponding
    /// `Event::NewPeer` isn't received. See also [Process for
    /// Connecting]
    /// (https://github.com/maidsafe/crust/blob/master/docs/connect.md) for
    /// details on handling of connect in different protocols.
    pub fn connect(&self,
                   our_connection_info: OurConnectionInfo,
                   their_connection_info: TheirConnectionInfo) {
        let their_id = their_connection_info.id;
        if their_id == self.id() {
            return;
        }

        if !unwrap_result!(self.connection_map.lock())
                .get(&their_id)
                .into_iter()
                .all(Vec::is_empty) {
            return;
        }

        let _ = self.expected_peers.lock().unwrap().insert(their_id);

        {
            let i = &their_connection_info.static_contact_info;
            self.mapping_context.add_simple_tcp_servers(i.tcp_mapper_servers.iter().cloned());
        }

        let event_tx = self.event_tx.clone();
        let connection_map = self.connection_map.clone();
        let our_public_key = self.our_keys.0.clone();
        let bootstrap_cache = self.bootstrap_cache.clone();
        let name_hash = self.name_hash;

        let (result_tx, result_rx) = mpsc::channel();

        let static_contact_info = their_connection_info.static_contact_info.clone();
        for tcp_addr in their_connection_info.static_contact_info.tcp_acceptors {
            let event_tx = event_tx.clone();
            let connection_map = connection_map.clone();
            let expected_peers = self.expected_peers.clone();
            let result_tx = result_tx.clone();
            let bootstrap_cache = bootstrap_cache.clone();
            let static_contact_info = static_contact_info.clone();
            let heart_beat_timeout = self.heart_beat_timeout.clone();
            let inactivity_timeout = self.inactivity_timeout.clone();

            let _ = thread!("Service::connect tcp direct", move || {
                match connection::connect_tcp_endpoint(tcp_addr,
                                                       heart_beat_timeout,
                                                       inactivity_timeout,
                                                       our_public_key,
                                                       event_tx,
                                                       connection_map,
                                                       Some(expected_peers),
                                                       Some(their_id),
                                                       name_hash) {
                    Err(err) => {
                        let err_msg = format!("Tcp direct connect failed: {}", err);
                        let err = io::Error::new(err.kind(), err_msg);
                        let _ = result_tx.send(Err(err));
                    }
                    Ok(()) => {
                        let _ = result_tx.send(Ok(()));
                        match unwrap_result!(bootstrap_cache.lock())
                                  .update_contacts(vec![static_contact_info], vec![]) {
                            Ok(()) => (),
                            Err(e) => warn!("Failed to update bootstrap cache: {}", e),
                        };
                    }
                }
            });
        }

        if let Some(tcp_socket) = our_connection_info.tcp_socket {
            let tcp_info = their_connection_info.tcp_info;
            let event_tx = event_tx.clone();
            let connection_map = connection_map.clone();
            let result_tx = result_tx.clone();
            let priv_tcp_info = our_connection_info.priv_tcp_info;
            let is_alive = self.is_alive.clone();
            let heart_beat_timeout = self.heart_beat_timeout.clone();
            let inactivity_timeout = self.inactivity_timeout.clone();

            let _ = thread!("Service::connect tcp rendezvous", move || {
                let res = tcp_punch_hole(tcp_socket, priv_tcp_info, tcp_info).result_log();

                if !is_alive.load(Ordering::Relaxed) {
                    return;
                }

                match res {
                    Ok(tcp_stream) => {
                        match connection::tcp_rendezvous_connect(connection_map,
                                                                 heart_beat_timeout,
                                                                 inactivity_timeout,
                                                                 event_tx,
                                                                 tcp_stream,
                                                                 their_id) {
                            Ok(()) => {
                                let _ = result_tx.send(Ok(()));
                            }
                            Err(err) => {
                                let err_msg = format!("Tcp rendezvous connect failed: {}", err);
                                let err = io::Error::new(err.kind(), err_msg);
                                let _ = result_tx.send(Err(err));
                            }
                        }
                    }
                    Err(err) => {
                        let err: io::Error = From::from(err);
                        let err_msg = format!("Tcp hole punching failed: {}", err);
                        let err = io::Error::new(err.kind(), err_msg);
                        let _ = result_tx.send(Err(err));
                    }
                };
            });
        }


        let _ = thread!("Service::connect aggregate results", move || {
            let mut errors: Vec<io::Error> = Vec::new();
            loop {
                match result_rx.recv() {
                    // Don't need to send a message here. Stupidly, the connect functions will have
                    // raised the event from deep within them.
                    Ok(Ok(())) => break,
                    Ok(Err(err)) => errors.push(err),
                    Err(mpsc::RecvError) => {
                        // All of the senders have hung up without sending us an Ok(()). They all
                        // must have failed to connect.
                        let len = errors.len();
                        let mut err_str: String = From::from("Connect failed. errors:");
                        for (i, e) in errors.into_iter().enumerate() {
                            let _ = write!(err_str, " ({} of {}) {}", i + 1, len, e);
                        }
                        let err = io::Error::new(io::ErrorKind::TimedOut, err_str);
                        let _ = event_tx.send(Event::NewPeer(Err(err), their_id));
                        return;
                    }
                }
            }
        });
    }

    /// Lookup a mapped udp socket based on result_token
    pub fn prepare_connection_info(&mut self, result_token: u32) {
        // FIXME: If the listeners are directly addressable (direct full cone or upnp mapped etc.
        // then our conact info is our static liseners
        // for udp we can map another socket, but use same local port if accessable/mapped
        // otherwise do following
        let our_static_contact_info = self.static_contact_info.clone();
        let event_tx = self.event_tx.clone();

        let mapping_context = self.mapping_context.clone();
        let our_pub_key = self.our_keys.0.clone();
        let _joiner = thread!("PrepareConnectionInfo", move || {
            let (tcp_socket, (our_priv_tcp_info, our_pub_tcp_info)) =
                match MappedTcpSocket::new(&mapping_context).result_log() {
                    Ok(MappedTcpSocket { socket, endpoints }) => {
                        (Some(socket), gen_rendezvous_info(endpoints))
                    }
                    Err(err) => {
                        let _ =
                            event_tx.send(Event::ConnectionInfoPrepared(ConnectionInfoResult {
                                result_token: result_token,
                                result: Err(From::from(err)),
                            }));
                        return;
                    }
                };

            let event = Event::ConnectionInfoPrepared(ConnectionInfoResult {
                result_token: result_token,
                result: Ok(OurConnectionInfo {
                    id: peer_id::new_id(our_pub_key),
                    tcp_info: our_pub_tcp_info,
                    priv_tcp_info: our_priv_tcp_info,
                    tcp_socket: tcp_socket,
                    static_contact_info: unwrap_result!(our_static_contact_info.lock()).clone(),
                }),
            });
            let _ = event_tx.send(event);
        });
    }

    /// Returns our ID.
    pub fn id(&self) -> PeerId {
        peer_id::new_id(self.our_keys.0)
    }
}

impl Drop for Service {
    fn drop(&mut self) {
        self.is_alive.store(false, Ordering::Relaxed);

        // Disconnect from all peers when we drop the service
        let mut cm = unwrap_result!(self.connection_map.lock());
        cm.clear();
    }
}

#[cfg(test)]
mod test {
    use super::*;
    use config_handler::Config;
    use event::Event;

    use std::time::Duration;
    use std::sync::atomic::{AtomicUsize, ATOMIC_USIZE_INIT, Ordering};
    use std::sync::{Arc, Barrier};
    use std::sync::mpsc;
    use std::sync::mpsc::Receiver;
    use std::thread;
    use std::thread::JoinHandle;
    use std::collections::{hash_map, HashMap};

    use crossbeam;
    use void::Void;
    use maidsafe_utilities::event_sender::{MaidSafeObserver, MaidSafeEventCategory};


    fn get_event_sender
        ()
        -> (::CrustEventSender, Receiver<MaidSafeEventCategory>, Receiver<Event>)
    {
        let (category_tx, category_rx) = mpsc::channel();
        let event_category = MaidSafeEventCategory::Crust;
        let (event_tx, event_rx) = mpsc::channel();

        (MaidSafeObserver::new(event_tx, event_category, category_tx), category_rx, event_rx)
    }

    fn timebomb<R, F>(dur: Duration, f: F) -> R
        where R: Send,
              F: Send + FnOnce() -> R
    {
        crossbeam::scope(|scope| {
            let thread_handle = thread::current();
            let (done_tx, done_rx) = mpsc::channel::<Void>();
            let jh = scope.spawn(move || {
                let ret = f();
                drop(done_tx);
                thread_handle.unpark();
                ret
            });
            thread::park_timeout(dur);
            match done_rx.try_recv() {
                Ok(x) => match x {},
                Err(mpsc::TryRecvError::Empty) => panic!("Timed out!"),
                Err(mpsc::TryRecvError::Disconnected) => jh.join(),
            }
        })
    }

    // Generate unique name for the bootstrap cache.
    fn gen_bootstrap_cache_name() -> String {
        static COUNTER: AtomicUsize = ATOMIC_USIZE_INIT;
        format!("test{}.bootstrap.cache",
                COUNTER.fetch_add(1, Ordering::Relaxed))
    }

    fn gen_beacon_port() -> u16 {
        const BASE: u16 = 40000;
        static COUNTER: AtomicUsize = ATOMIC_USIZE_INIT;

        BASE + COUNTER.fetch_add(1, Ordering::Relaxed) as u16
    }

    fn gen_config() -> Config {
        let mut config = Config::default();
        config.bootstrap_cache_name = Some(gen_bootstrap_cache_name());
        config
    }

    fn gen_config_with_beacon(port: u16) -> Config {
        let mut config = gen_config();
        config.service_discovery_port = Some(port);
        config
    }


    fn prepare_connection_info(service: &mut Service,
                               event_rx: &Receiver<Event>)
                               -> OurConnectionInfo {
        static TOKEN_COUNTER: AtomicUsize = ATOMIC_USIZE_INIT;
        let token = TOKEN_COUNTER.fetch_add(1, Ordering::Relaxed) as u32;

        service.prepare_connection_info(token);

        match unwrap_result!(event_rx.recv()) {
            Event::ConnectionInfoPrepared(cir) => {
                assert_eq!(cir.result_token, token);
                unwrap_result!(cir.result)
            }
            event => panic!("Received unexpected event: {:?}", event),
        }
    }

    #[test]
    fn start_stop_service() {
        let config = gen_config();
        let (event_sender, _category_rx, _) = get_event_sender();
        let _service = unwrap_result!(Service::with_config(event_sender, &config));
    }

    #[test]
    fn bootstrap_connection_two_services() {
        let (event_sender_0, _category_rx_0, event_rx_0) = get_event_sender();
        let (event_sender_1, _category_rx_1, event_rx_1) = get_event_sender();

        let beacon_port = gen_beacon_port();
        let config_0 = gen_config_with_beacon(beacon_port);
        let config_1 = gen_config_with_beacon(beacon_port);

        let mut service_0 = unwrap_result!(Service::with_config(event_sender_0, &config_0));
        let _ = service_0.start_listening_tcp();

        // let service_0 finish bootstrap - since it is the zero state, it should not find any peer
        // to bootstrap
        {
            let event_rxd = unwrap_result!(event_rx_0.recv());
            match event_rxd {
                Event::BootstrapFinished => (),
                _ => panic!("Received unexpected event: {:?}", event_rxd),
            }
        }
        service_0.start_service_discovery();

        let mut service_1 = unwrap_result!(Service::with_config(event_sender_1, &config_1));
        let _ = service_1.start_listening_tcp();

        // let service_1 finish bootstrap - it should bootstrap off service_0
        let id_0 = {
            let event_rxd = unwrap_result!(event_rx_1.recv());
            match event_rxd {
                Event::BootstrapConnect(their_id) => their_id,
                _ => panic!("Received unexpected event: {:?}", event_rxd),
            }
        };

        // now service_1 should get BootstrapFinished
        {
            let event_rxd = unwrap_result!(event_rx_1.recv());
            match event_rxd {
                Event::BootstrapFinished => (),
                _ => panic!("Received unexpected event: {:?}", event_rxd),
            }
        }

        // service_0 should have received service_1's bootstrap connection by now
        let id_1 = match unwrap_result!(event_rx_0.recv()) {
            Event::BootstrapAccept(their_id) => their_id,
            _ => panic!("0 Should have got a new connection from 1."),
        };



        assert!(id_0 != id_1);

        // send data from 0 to 1
        {
            let data_txd = vec![0, 1, 255, 254, 222, 1];
            unwrap_result!(service_0.send(&id_1, data_txd.clone(), 0));

            // 1 should rx data
            let (data_rxd, peer_id) = {
                let event_rxd = unwrap_result!(event_rx_1.recv());
                match event_rxd {
                    Event::NewMessage(their_id, msg) => (msg, their_id),
                    _ => panic!("Received unexpected event: {:?}", event_rxd),
                }
            };

            assert_eq!(data_rxd, data_txd);
            assert_eq!(peer_id, id_0);
        }

        // send data from 1 to 0
        {
            let data_txd = vec![10, 11, 155, 214, 202];
            unwrap_result!(service_1.send(&id_0, data_txd.clone(), 0));

            // 0 should rx data
            let (data_rxd, peer_id) = {
                let event_rxd = unwrap_result!(event_rx_0.recv());
                match event_rxd {
                    Event::NewMessage(their_id, msg) => (msg, their_id),
                    _ => panic!("Received unexpected event: {:?}", event_rxd),
                }
            };

            assert_eq!(data_rxd, data_txd);
            assert_eq!(peer_id, id_1);
        }

        assert!(service_0.disconnect(&id_1));

        match unwrap_result!(event_rx_1.recv()) {
            Event::LostPeer(id) => assert_eq!(id, id_0),
            e => panic!("Received unexpected event: {:?}", e),
        }

        // Drop services and make sure the event channels close
        drop(service_0);
        let (done_tx, done_rx) = mpsc::channel();
        let thread_handle = thread::current();
        let tj = thread!("Drain event channel messages", move || {
            match event_rx_0.recv() {
                Ok(e) => panic!("Received unexpected event when shutting down: {:?}", e),
                Err(mpsc::RecvError) => (),
            };
            unwrap_result!(done_tx.send(()));
            thread_handle.unpark();
        });
        thread::park_timeout(Duration::from_secs(60));
        unwrap_result!(done_rx.try_recv());
        unwrap_result!(tj.join());

        drop(service_1);
        let (done_tx, done_rx) = mpsc::channel();
        let thread_handle = thread::current();
        let tj = thread!("Drain event channel messages", move || {
            match event_rx_1.recv() {
                Ok(e) => panic!("Received unexpected event when shutting down: {:?}", e),
                Err(mpsc::RecvError) => (),
            };
            unwrap_result!(done_tx.send(()));
            thread_handle.unpark();
        });
        thread::park_timeout(Duration::from_secs(5));
        unwrap_result!(done_rx.try_recv());
        unwrap_result!(tj.join());
    }

    #[test]
    fn peer_to_peer_connection_two_services() {
        let (event_sender_0, _category_rx_0, event_rx_0) = get_event_sender();
        let (event_sender_1, _category_rx_1, event_rx_1) = get_event_sender();

        let config = gen_config();

        let mut service_0 = unwrap_result!(Service::with_config(event_sender_0, &config));
        let _ = service_0.start_listening_tcp();

        // let service_0 finish bootstrap - since it is the zero state, it should not find any peer
        // to bootstrap
        match unwrap_result!(event_rx_0.recv()) {
            Event::BootstrapFinished => (),
            event => panic!("Received unexpected event: {:?}", event),
        }

        let mut service_1 = unwrap_result!(Service::with_config(event_sender_1, &config));
        let _ = service_1.start_listening_tcp();

        // let service_0 finish bootstrap - since it is the zero state, it should not find any peer
        // to bootstrap
        match unwrap_result!(event_rx_1.recv()) {
            Event::BootstrapFinished => (),
            event => panic!("Received unexpected event: {:?}", event),
        }

        let our_ci_0 = prepare_connection_info(&mut service_0, &event_rx_0);
        let their_ci_0 = our_ci_0.to_their_connection_info();

        let our_ci_1 = prepare_connection_info(&mut service_1, &event_rx_1);
        let their_ci_1 = our_ci_1.to_their_connection_info();

        service_0.connect(our_ci_0, their_ci_1);
        service_1.connect(our_ci_1, their_ci_0);

        let id_1 = match unwrap_result!(event_rx_0.recv()) {
            Event::NewPeer(Ok(()), their_id) => their_id,
            m => panic!("0 Should have connected to 1. Got message {:?}", m),
        };

        let id_0 = match unwrap_result!(event_rx_1.recv()) {
            Event::NewPeer(Ok(()), their_id) => their_id,
            m => panic!("1 Should have connected to 0. Got message {:?}", m),
        };

        // send data from 0 to 1
        {
            let data_txd = vec![0, 1, 255, 254, 222, 1];
            unwrap_result!(service_0.send(&id_1, data_txd.clone(), 0));

            // 1 should rx data
            let (data_rxd, peer_id) = {
                let event_rxd = unwrap_result!(event_rx_1.recv());
                match event_rxd {
                    Event::NewMessage(their_id, msg) => (msg, their_id),
                    _ => panic!("Received unexpected event: {:?}", event_rxd),
                }
            };

            assert_eq!(data_rxd, data_txd);
            assert_eq!(peer_id, id_0);
        }

        // send data from 1 to 0
        {
            let data_txd = vec![10, 11, 155, 214, 202];
            unwrap_result!(service_1.send(&id_0, data_txd.clone(), 0));

            // 0 should rx data
            let (data_rxd, peer_id) = {
                let event_rxd = unwrap_result!(event_rx_0.recv());
                match event_rxd {
                    Event::NewMessage(their_id, msg) => (msg, their_id),
                    _ => panic!("Received unexpected event: {:?}", event_rxd),
                }
            };

            assert_eq!(data_rxd, data_txd);
            assert_eq!(peer_id, id_1);
        }

        // Drop services and make sure the event channels close
        drop(service_0);
        match event_rx_1.recv() {
            Ok(Event::LostPeer(id)) => assert_eq!(id_0, id),
            event => panic!("Received unexpected event: {:?}", event),
        }
    }


    // TODO: change this to allow arbitrary number of nodes.
    #[test]
    fn rendezvous_connection_three_services() {
        const NUM_SERVICES: usize = 3;
        const MSG_SIZE: usize = 1024;
        const NUM_MSGS: usize = 257;

        struct TestNode {
            event_rx: Receiver<Event>,
            _category_rx: Receiver<MaidSafeEventCategory>,
            service: Service,
            connection_id_rx: Receiver<TheirConnectionInfo>,
            our_cis: Vec<OurConnectionInfo>,
            our_index: usize,
        }

        impl TestNode {
            fn new(index: usize) -> (TestNode, mpsc::Sender<TheirConnectionInfo>) {
                let (event_sender, category_rx, event_rx) = get_event_sender();
                let config = unwrap_result!(::config_handler::read_config_file());
                let service = unwrap_result!(Service::with_config(event_sender, &config));
                match unwrap_result!(event_rx.recv()) {
                    Event::BootstrapFinished => (),
                    m => panic!("Expected BootstrapFinished, got:{:?}", m),
                };
                let (ci_tx, ci_rx) = mpsc::channel();
                (TestNode {
                    event_rx: event_rx,
                    _category_rx: category_rx,
                    service: service,
                    connection_id_rx: ci_rx,
                    our_cis: Vec::new(),
                    our_index: index,
                },
                 ci_tx)
            }

            fn make_connection_infos(&mut self, ci_txs: &[mpsc::Sender<TheirConnectionInfo>]) {
                for (i, ci_tx) in ci_txs.iter().enumerate() {
                    if i == self.our_index {
                        continue;
                    }

                    let our_ci = prepare_connection_info(&mut self.service, &self.event_rx);
                    let their_ci = our_ci.to_their_connection_info();
                    let _ = ci_tx.send(their_ci);
                    self.our_cis.push(our_ci);
                }
            }

            fn run(self, send_barrier: Arc<Barrier>, drop_barrier: Arc<Barrier>) -> JoinHandle<()> {
                thread!("run!", move || {
                    for (our_ci, their_ci) in self.our_cis
                                                  .into_iter()
                                                  .zip(self.connection_id_rx.into_iter()) {
                        self.service.connect(our_ci, their_ci);
                    }
                    let mut their_ids = HashMap::new();
                    for _ in 0..(NUM_SERVICES - 1) {
                        let their_id = match unwrap_result!(self.event_rx.recv()) {
                            Event::NewPeer(Ok(()), their_id) => their_id,
                            m => panic!("Expected NewPeer message. Got message {:?}", m),
                        };
                        match their_ids.insert(their_id, 0u32) {
                            Some(_) => panic!("Received two NewPeer events for same peer!"),
                            None => (),
                        };
                    }

                    // Wait until all nodes have connected to each other before we start
                    // exchanging messages.
                    let _ = send_barrier.wait();

                    for their_id in their_ids.keys() {
                        for n in 0..NUM_MSGS {
                            let mut msg = Vec::with_capacity(MSG_SIZE);
                            for _ in 0..MSG_SIZE {
                                msg.push(n as u8);
                            }
                            let _ = self.service.send(their_id, msg, 0);
                        }
                    }

                    for _ in 0..((NUM_SERVICES - 1) * NUM_MSGS) {
                        match unwrap_result!(self.event_rx.recv()) {
                            Event::NewMessage(their_id, msg) => {
                                let n = msg[0];
                                assert_eq!(msg.len(), MSG_SIZE);
                                for m in msg {
                                    assert_eq!(n, m);
                                }
                                match their_ids.entry(their_id.clone()) {
                                    hash_map::Entry::Occupied(mut oe) => {
                                        let next_msg = oe.get_mut();
                                        assert_eq!(*next_msg as u8, n);
                                        *next_msg += 1;
                                    }
                                    hash_map::Entry::Vacant(_) => panic!("impossible!"),
                                }
                            }
                            m => panic!("Unexpected msg receiving NewMessage: {:?}", m),
                        }
                    }

                    // Wait until all nodes have finished exchanging messages before we start
                    // disconnecting.
                    let _ = drop_barrier.wait();

                    drop(self.service);
                    match self.event_rx.recv() {
                        Ok(m) => {
                            match m {
                                Event::LostPeer(..) => (),
                                _ => panic!("Unexpected message when shutting down: {:?}", m),
                            }
                        }
                        Err(mpsc::RecvError) => (),
                    }
                })
            }
        }

        let mut test_nodes = Vec::new();
        let mut ci_txs = Vec::new();
        for i in 0..NUM_SERVICES {
            let (test_node, ci_tx) = TestNode::new(i);
            test_nodes.push(test_node);
            ci_txs.push(ci_tx);
        }

        for test_node in &mut test_nodes {
            test_node.make_connection_infos(&ci_txs);
        }

        let send_barrier = Arc::new(Barrier::new(NUM_SERVICES));
        let drop_barrier = Arc::new(Barrier::new(NUM_SERVICES));
        let mut threads = Vec::new();
        for test_node in test_nodes {
            let send_barrier = send_barrier.clone();
            let drop_barrier = drop_barrier.clone();
            threads.push(test_node.run(send_barrier, drop_barrier));
        }

        // Wait one hundred millisecond per message
        // TODO(canndrew): drop this limit
        let timeout_ms = 100 * (NUM_MSGS * (NUM_SERVICES * (NUM_SERVICES - 1)) / 2) as u64;
        timebomb(Duration::from_millis(timeout_ms), move || {
            for thread in threads {
                unwrap_result!(thread.join());
            }
        });
    }


    #[test]
    fn drop_disconnects() {
        let beacon_port = gen_beacon_port();

        let config_0 = gen_config_with_beacon(beacon_port);
        let config_1 = gen_config_with_beacon(beacon_port);

        let (event_sender_0, _category_rx_0, event_rx_0) = get_event_sender();
        let (event_sender_1, _category_rx_1, event_rx_1) = get_event_sender();

        let mut service_0 = unwrap_result!(Service::with_config(event_sender_0, &config_0));
        unwrap_result!(service_0.start_listening_tcp());

        // Let service_0 finish bootstrap - it should not find any peer.
        match unwrap_result!(event_rx_0.recv()) {
            Event::BootstrapFinished => (),
            event_rxd => panic!("Received unexpected event: {:?}", event_rxd),
        }
        service_0.start_service_discovery();

        let mut service_1 = unwrap_result!(Service::with_config(event_sender_1, &config_1));
        unwrap_result!(service_1.start_listening_tcp());

        // Let service_1 finish bootstrap - it should bootstrap off service_0.
        let id_0 = match unwrap_result!(event_rx_1.recv()) {
            Event::BootstrapConnect(their_id) => their_id,
            event => panic!("Received unexpected event: {:?}", event),
        };

        // Now service_1 should get BootstrapFinished.
        match unwrap_result!(event_rx_1.recv()) {
            Event::BootstrapFinished => (),
            event => panic!("Received unexpected event: {:?}", event),
        }

        // service_0 should have received service_1's bootstrap connection by now.
        let _ = match unwrap_result!(event_rx_0.recv()) {
            Event::BootstrapAccept(their_id) => their_id,
            _ => panic!("0 Should have got a new connection from 1."),
        };

        // Dropping service_0 should make service_1 receive a LostPeer event.
        drop(service_0);
        match unwrap_result!(event_rx_1.recv()) {
            Event::LostPeer(id) => assert_eq!(id, id_0),
            event => panic!("Received unexpected event: {:?}", event),
        }
    }

    #[test]
    fn skip_invalid_bootstrap_contacts() {
        use config_handler::Config;
        use socket_addr::SocketAddr;
        use static_contact_info::StaticContactInfo;
        use std::net;
        use std::str::FromStr;

        let mut contact_info = StaticContactInfo::default();

        let invalid_addrs = vec![
            SocketAddr(net::SocketAddr::from_str("127.0.0.1:0").unwrap()),
        ];

        contact_info.tcp_acceptors = invalid_addrs.clone();

        let mut config = Config::default();
        config.hard_coded_contacts = vec![contact_info];

        let (event_sender, _category_rx, event_rx) = get_event_sender();
        let _service = unwrap_result!(Service::with_config(event_sender, &config));

        timebomb(Duration::from_secs(70), move || {
            match unwrap_result!(event_rx.recv()) {
                Event::BootstrapFinished => (),
                event => panic!("Received unexpected event: {:?}", event),
            }
        });
    }

    #[test]
    fn new_peer_is_not_raised_if_only_one_party_calls_connect() {
        let config_0 = gen_config();
        let config_1 = gen_config();

        let (event_sender_0, _category_rx_0, event_rx_0) = get_event_sender();
        let (event_sender_1, _category_rx_1, event_rx_1) = get_event_sender();

        let mut service_0 = unwrap_result!(Service::with_config(event_sender_0, &config_0));
        match unwrap_result!(event_rx_0.recv()) {
            Event::BootstrapFinished => (),
            event => panic!("Received unexpected event: {:?}", event),
        }

        unwrap_result!(service_0.start_listening_tcp());

        let mut service_1 = unwrap_result!(Service::with_config(event_sender_1, &config_1));
        match unwrap_result!(event_rx_1.recv()) {
            Event::BootstrapFinished => (),
            event => panic!("Received unexpected event: {:?}", event),
        }

        unwrap_result!(service_1.start_listening_tcp());

        let our_ci_0 = prepare_connection_info(&mut service_0, &event_rx_0);
        let our_ci_1 = prepare_connection_info(&mut service_1, &event_rx_1);
        let their_ci_1 = our_ci_1.to_their_connection_info();

        service_0.connect(our_ci_0, their_ci_1);

        thread::sleep(Duration::from_millis(1000));

        match event_rx_0.try_recv() {
            Ok(Event::NewPeer(Ok(()), _)) => panic!("Unexpected NewPeer event"),
            _ => (),
        }

        match event_rx_1.try_recv() {
            Ok(Event::NewPeer(Ok(()), _)) => panic!("Unexpected NewPeer event"),
            _ => (),
        }
    }
}
