// Copyright 2020 Netwarps Ltd.
//
// Permission is hereby granted, free of charge, to any person obtaining a
// copy of this software and associated documentation files (the "Software"),
// to deal in the Software without restriction, including without limitation
// the rights to use, copy, modify, merge, publish, distribute, sublicense,
// and/or sell copies of the Software, and to permit persons to whom the
// Software is furnished to do so, subject to the following conditions:
//
// The above copyright notice and this permission notice shall be included in
// all copies or substantial portions of the Software.
//
// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS
// OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING
// FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER
// DEALINGS IN THE SOFTWARE.

//! High level manager of the network.
//!
//! A [`Swarm`] contains the state of the network as a whole. The entire
//! behaviour of a libp2p network can be controlled through the `Swarm`.
//! The `Swarm` struct contains all established connections to remotes and
//! manages the state of all the substreams that have been opened, and all
//! the upgrades that were built upon these substreams.
//!
//! # Initializing a Swarm
//!
//! Creating a `Swarm` requires three things:
//!
//!  1. A network identity of the local node in form of a [`PeerId`].
//!  2. One or more implementations of the [`Transport`] trait. This is the
//!     type that will be used in order to reach nodes on the network based
//!     on their address. See the `transport` module for more information.
//!  3. One or more implementations of the [`ProtocolHandler`] trait. This is
//!     the protocols that `Swarm` is going to support.
//!
//! # Protocol Handler
//!
//! The [`ProtocolHandler`] trait defines how each active connection to a
//! remote should behave: how to handle incoming substreams, which protocols
//! are supported, etc.
//!

pub mod connection;
mod control;
mod dial;
mod muxer;
pub mod network;
mod registry;

pub mod cli;
pub mod identify;
pub mod metrics;
pub mod ping;
pub mod protocol_handler;
pub mod substream;

pub use control::Control;
pub use protocol_handler::DummyProtocolHandler;

use fnv::FnvHashMap;
use futures::channel::{mpsc, oneshot};
use futures::future::Either;
use futures::prelude::*;
use smallvec::SmallVec;
use std::collections::HashSet;
use std::sync::Arc;
use std::time::Duration;
use std::{error, fmt};

use async_std::task;

use libp2prs_core::peerstore::{PeerStore, ADDRESS_TTL};
use libp2prs_core::{
    multiaddr::{protocol, Multiaddr},
    muxing::IStreamMuxer,
    transport::{upgrade::ITransportEx, TransportError},
    PeerId, ProtocolId, PublicKey,
};

use crate::connection::{Connection, ConnectionId, ConnectionLimit, ConnectionView, Direction};
use crate::control::{DumpCommand, SwarmControlCmd};
use crate::dial::EitherDialAddr;
use crate::identify::{IdentifyConfig, IdentifyHandler, IdentifyInfo, IdentifyPushHandler};
use crate::metrics::metric::Metric;
use crate::muxer::Muxer;
use crate::network::NetworkInfo;
use crate::ping::{PingConfig, PingHandler};
use crate::protocol_handler::IProtocolHandler;
use crate::registry::Addresses;
use crate::substream::{ConnectInfo, StreamId, Substream, SubstreamView};
use libp2prs_core::routing::IRouting;
use libp2prs_core::translation::address_translation;
use libp2prs_core::transport::ListenerEvent;

type Result<T> = std::result::Result<T, SwarmError>;

pub const PEERSTORE_GC_PURGE_INTERVAL: Duration = Duration::from_secs(10 * 60);

const LIBP2P_RS_PROTOCOL_VERSION: &str = "ipfs/0.1.0";
const LIBP2P_RS_AGENT_VERSION: &str = "libp2p-rs/0.1.0";

/// Event generated by the `Swarm`.
#[derive(Debug)]
pub enum SwarmEvent {
    /// A connection to the given peer has been opened.
    ConnectionEstablished {
        /// The connection, AKA. the trait object of stream muxer.
        stream_muxer: IStreamMuxer,
        /// Direction of the connection.
        direction: Direction,
        /// The dial transaction id
        tid: Option<TransactionId>,
    },
    /// A connection with the given peer has been closed.
    ConnectionClosed {
        /// The connection Id of the sub stream.
        cid: ConnectionId,
        /// The cause of the error.
        error: TransportError,
    },
    /// A stream failed to negotiate protocol with peer.
    StreamError {
        /// The connection Id of the sub stream.
        cid: ConnectionId,
        /// The cause of the error.
        error: TransportError,
    },
    /// A new substream opened.
    StreamOpened {
        /// The view of the opened substream.
        view: SubstreamView,
    },
    /// An error happened on accepting incoming connection.
    ///
    /// This can include, for example, an error during the handshake of the encryption layer, or
    /// the connection unexpectedly closed.
    IncomingConnectionError {
        /// The remote multiaddr.
        remote_addr: Multiaddr,
        // The error that happened when listening.
        error: TransportError,
    },
    OutgoingConnectionError {
        /// The remote Peer Id.
        peer_id: PeerId,
        // /// The remote multiaddr.
        // remote_addr: Multiaddr,
        /// The error that happened when dialing.
        error: SwarmError,
        // The dial transaction id
        tid: TransactionId,
    },
    /// One of our listeners has a new address added.
    ListenAddressAdded(Multiaddr),
    /// One of our listeners has an address deleted.
    ListenAddressDeleted(Multiaddr),
    /// One of the listeners gracefully closed.
    ///
    /// Seems unlikely, as least in async-std
    ListenerClosed {
        /// The addresses that the listener was listening on.
        addresses: Vec<Multiaddr>,
        /// Reason for the closure. Contains `Ok(())` if the stream produced `None`, or `Err`
        /// if the stream produced an error.
        reason: TransportError,
    },

    /// A Ping result generated by the Ping protocol handler.
    PingResult {
        /// The connection Id.
        cid: ConnectionId,
        /// The result.
        /// Duration means the TTL when succeeded, or SwarmError for failed.
        result: Result<Duration>,
    },
    /// An identify result generated by the Identify protocol handler.
    IdentifyResult {
        /// The connection Id.
        cid: ConnectionId,
        /// The result.
        /// Duration means the TTL when succeeded, or SwarmError for failed.
        result: Result<(IdentifyInfo, Multiaddr)>,
    },
}

/// The Transports helper.
#[derive(Clone, Default)]
struct Transports {
    inner: FnvHashMap<u32, ITransportEx>,
}

impl Transports {
    pub(crate) fn lookup_by_addr(&self, mut addr: Multiaddr) -> Result<ITransportEx> {
        log::debug!("lookup transport for addr={}", addr);
        if let Some(d) = addr.pop() {
            if let Ok(id) = d.get_key() {
                if let Some(transport) = self.inner.get(&id).map(|s| s.box_clone()) {
                    return Ok(transport);
                }
            }
        }
        Err(SwarmError::Transport(TransportError::MultiaddrNotSupported(addr)))
    }
    pub(crate) fn get(&self, id: &u32) -> Option<&ITransportEx> {
        self.inner.get(id)
    }
    pub(crate) fn add(&mut self, id: u32, transport: ITransportEx) -> Option<ITransportEx> {
        self.inner.insert(id, transport)
    }
}

/// The post-processing callback for Dialer.
type DialCallback = Box<dyn FnOnce(Result<&mut Connection>) + Send>;
/// The transaction Id for dialer post-processing.
type TransactionId = usize;

/// Contains the state of the network, plus the way it should behave.
pub struct Swarm {
    // to improve peerstore... we don't want to leak PeerStore
    peer_store: PeerStore,

    /// The protocol multistream selector.
    muxer: Muxer,

    /// The routing interface.
    routing: Option<IRouting>,

    /// The Transports.
    transports: Transports,

    /// The public key
    public_key: PublicKey,
    /// The local peer ID.
    local_peer_id: PeerId,

    /// The next listener ID to assign.
    next_connection_id: usize,

    /// List of multiaddresses we're listening on.
    listened_addrs: SmallVec<[Multiaddr; 8]>,

    /// List of multiaddresses we're listening on, after account for external IP addresses and
    /// similar mechanisms.
    external_addrs: Addresses,

    /// Metrics. Monitor the network resource that spend on connection
    metric: Arc<Metric>,

    /// List of nodes for which are forbidden.
    banned_peers: HashSet<PeerId>,

    /// The all connections_by_peer connections organized by their Ids
    connections_by_id: FnvHashMap<ConnectionId, Connection>,
    /// The all connections_by_peer connections by  peer Id
    /// There might be more than one connections for a remote peer
    connections_by_peer: FnvHashMap<PeerId, Vec<ConnectionId>>,

    /// Swarm will listen on this channel, waiting for events generated from underlying transport
    event_receiver: mpsc::UnboundedReceiver<SwarmEvent>,
    /// The Swarm event sender wil be cloned and then taken by underlying parts
    event_sender: mpsc::UnboundedSender<SwarmEvent>,

    /// Swarm will listen on this channel, for external control commands
    ctrl_receiver: mpsc::Receiver<SwarmControlCmd>,
    /// The Swarm event sender wil be cloned and then taken by others
    ctrl_sender: mpsc::Sender<SwarmControlCmd>,

    /// The dialer for making outgoing connections.
    dialer: dial::AsyncDialer,

    /// The dialer unique transaction id.
    next_tid: TransactionId,
    /// The dialer post-processing hashmap.
    dial_transactions: FnvHashMap<TransactionId, DialCallback>,
}

#[allow(dead_code)]
impl Swarm {
    /// Builds a new `Swarm`.
    pub fn new(
        key: PublicKey,
        //_config: NetworkConfig,
    ) -> Self {
        // unbounded channel for events, so that we can send a message to ourselves
        let (event_tx, event_rx) = mpsc::unbounded();
        let (ctrl_tx, ctrl_rx) = mpsc::channel(0);

        let peer_store = PeerStore::default();
        // peer_store.add_key(&key.clone().into_peer_id(), key.clone());

        if let Err(e) = peer_store.load_data() {
            if e.kind() != std::io::ErrorKind::UnexpectedEof {
                log::info!("PeerStore load data error: {}", e);
            }
        }

        let metric = Metric::new();

        Swarm {
            peer_store,
            muxer: Muxer::new(),
            routing: None,
            transports: Default::default(),
            public_key: key.clone(),
            local_peer_id: key.into_peer_id(),
            next_connection_id: 0,
            listened_addrs: Default::default(),
            external_addrs: Default::default(),
            banned_peers: Default::default(),
            connections_by_id: Default::default(),
            connections_by_peer: Default::default(),
            metric: Arc::new(metric),
            event_receiver: event_rx,
            event_sender: event_tx,
            ctrl_receiver: ctrl_rx,
            ctrl_sender: ctrl_tx,
            dialer: dial::AsyncDialer::new(),
            next_tid: 0,
            dial_transactions: Default::default(),
        }
    }
    fn assign_cid(&mut self) -> usize {
        self.next_connection_id += 1;
        self.next_connection_id
    }
    fn assign_tid(&mut self) -> TransactionId {
        self.next_tid += 1;
        self.next_tid
    }

    /// Adds transport to Swarm.
    pub fn with_transport(mut self, transport: ITransportEx) -> Self {
        let protocols = transport.protocols();
        if protocols.is_empty() {
            panic!("Shouldn't happen: no protocols found in Transport");
        }
        let mut registered: Vec<protocol::Protocol> = vec![];
        for p in protocols.iter() {
            if self.transports.get(p).is_some() {
                let proto = protocol::Protocol::get_enum(*p).unwrap();
                registered.push(proto);
            }
        }
        if !registered.is_empty() {
            log::info!("transports already registered for protocol(s): {:?}", registered);
        }

        for p in protocols.iter() {
            log::debug!("add protocol={}", p);
            self.transports.add(*p, transport.box_clone());
        }
        self
    }
    /// Creates Swarm with protocol handler.
    pub fn with_protocol(mut self, p: IProtocolHandler) -> Self {
        self.muxer.add_protocol_handler(p);
        self
    }
    /// Creates Swarm with Ping service.
    pub fn with_ping(mut self, ping: PingConfig) -> Self {
        self.muxer.add_protocol_handler(Box::new(PingHandler::new(ping)));
        self
    }
    /// Creates Swarm with routing service.
    pub fn with_routing(mut self, routing: IRouting) -> Self {
        self.routing = Some(routing);
        self
    }
    /// Creates Swarm with Metrics.
    pub fn with_metric(mut self, metric: Metric) -> Self {
        self.metric = Arc::new(metric);
        self
    }
    /// Creates Swarm with Identify service.
    pub fn with_identify(mut self, config: IdentifyConfig) -> Self {
        let handler = IdentifyHandler::new(self.ctrl_sender.clone());
        self.muxer.add_protocol_handler(Box::new(handler));
        let handler = IdentifyPushHandler::new(config, self.event_sender.clone());
        self.muxer.add_protocol_handler(Box::new(handler));
        self
    }

    /// Get an API controller for Swarm.
    pub fn control(&self) -> Control {
        Control::new(self.ctrl_sender.clone(), self.peer_store.clone(), self.metric.clone())
    }

    /// Handles events generated internally or externally.
    async fn handle_messages(&mut self) -> Result<()> {
        loop {
            let either = future::select(self.event_receiver.next(), self.ctrl_receiver.next()).await;
            match either {
                Either::Left((evt, _)) => {
                    if let Some(evt) = evt {
                        self.on_event(evt);
                    } else {
                        // we are closed anyway, break
                        log::debug!("Swarm event channel is closed, closing down...");
                        return Err(SwarmError::Closing(2));
                    }
                }
                Either::Right((cmd, _)) => {
                    if let Some(cmd) = cmd {
                        let _ = self.on_command(cmd);
                    } else {
                        return Err(SwarmError::Closing(1));
                    }
                }
            }
        }
    }

    // to kick off address change event to all protocol handlers
    fn kickoff_address_change(&mut self) {
        // kick off all protocol handlers for the Address Change
        let own_addrs = self.get_self_addrs();
        for handler in self.muxer.protocol_handlers.values_mut() {
            handler.address_changed(own_addrs.clone());
        }
    }

    fn on_event(&mut self, event: SwarmEvent) {
        log::trace!("Swarm event={:?}", event);
        match event {
            SwarmEvent::ListenerClosed { addresses: _, reason: _ } => {}
            SwarmEvent::ListenAddressAdded(addr) => {
                if !self.listened_addrs.contains(&addr) {
                    log::info!("New address added {}", addr);
                    self.listened_addrs.push(addr);
                    self.kickoff_address_change();
                }
            }
            SwarmEvent::ListenAddressDeleted(addr) => {
                if let Some(pos) = self.listened_addrs.iter().position(|a| a == &addr) {
                    log::info!("Old address deleted {}", addr);
                    self.listened_addrs.remove(pos);
                    self.kickoff_address_change();
                }
            }
            SwarmEvent::ConnectionEstablished {
                stream_muxer,
                direction,
                tid,
            } => {
                let _ = self.handle_connection_opened(stream_muxer, direction, tid);
            }
            SwarmEvent::ConnectionClosed { cid, error: _ } => {
                let _ = self.handle_connection_closed(cid);
            }
            SwarmEvent::OutgoingConnectionError { tid, peer_id, error } => {
                let _ = self.handle_connection_error(peer_id, error, tid);
            }
            SwarmEvent::IncomingConnectionError { remote_addr, error } => {
                log::debug!("incoming connection error for {:?} {:?}", remote_addr, error);
            }
            SwarmEvent::StreamError { .. } => {
                // TODO: add statistics
            }
            SwarmEvent::StreamOpened { view } => {
                let _ = self.handle_stream_opened(view);
            }
            SwarmEvent::PingResult { cid, result } => {
                let _ = self.handle_ping_result(cid, result);
            }
            SwarmEvent::IdentifyResult { cid, result } => {
                let _ = self.handle_identify_result(cid, result);
            }
        }
    }

    fn on_command(&mut self, cmd: SwarmControlCmd) -> Result<()> {
        log::trace!("Swarm control command={:?}", cmd);

        match cmd {
            SwarmControlCmd::Connect(peer_id, addrs, reply) => {
                // got the peer_id and the addresses, start the dialer for it
                let _ = self.on_connect(peer_id, addrs, reply);
            }
            SwarmControlCmd::NewConnection(peer_id, use_routing, reply) => {
                // got the peer_id, start the dialer for it
                let _ = self.on_new_connection(peer_id, use_routing, reply);
            }
            SwarmControlCmd::CloseConnection(peer_id, reply) => {
                // got the peer_id, close all connections to the peer
                let _ = self.on_close_connection(peer_id, reply);
            }
            SwarmControlCmd::NewStream(peer_id, pids, use_routing, reply) => {
                // got the peer_id, try opening a new sub stream
                let _ = self.on_new_stream(peer_id, pids, use_routing, reply);
            }
            SwarmControlCmd::CloseStream(cid, sid) => {
                // got the connection_id, try closing a new sub stream
                let _ = self.on_close_stream(cid, sid);
            }
            SwarmControlCmd::SelfAddresses(reply) => {
                // Received from channel, try retrieving identify info
                let _ = self.on_retrieve_own_addresses(|r| {
                    let _ = reply.send(r);
                });
            }
            SwarmControlCmd::NetworkInfo(reply) => {
                // Received from channel, try retrieving network info
                let _ = self.on_retrieve_network_info(|r| {
                    let _ = reply.send(r);
                });
            }
            SwarmControlCmd::IdentifyInfo(reply) => {
                // Received from channel, try retrieving identify info
                let _ = self.on_retrieve_identify_info(|r| {
                    let _ = reply.send(r);
                });
            }
            SwarmControlCmd::Dump(cmd) => match cmd {
                DumpCommand::Connections(peer_id, reply) => {
                    let _ = self.on_retrieve_connection_views(peer_id, |r| {
                        let _ = reply.send(r);
                    });
                }
                DumpCommand::Streams(peer_id, reply) => {
                    let _ = self.on_retrieve_substream_views(peer_id, |r| {
                        let _ = reply.send(r);
                    });
                }
            },
        }
        Ok(())
    }

    fn on_connect(&mut self, peer_id: PeerId, addrs: Vec<Multiaddr>, reply: oneshot::Sender<Result<()>>) -> Result<()> {
        // return if we already have the connection, otherwise, start dialing
        if let Some(_conn) = self.get_best_conn(&peer_id) {
            let _ = reply.send(Ok(()));
        } else {
            // dialing peer, with the post-processing callback
            self.dial_addr(peer_id, addrs, |r: Result<&mut Connection>| {
                let _ = reply.send(r.map(|_| ()));
            });
        }
        Ok(())
    }

    fn on_new_connection(&mut self, peer_id: PeerId, use_routing: bool, reply: oneshot::Sender<Result<()>>) -> Result<()> {
        // return if we already have the connection, otherwise, start dialing
        if let Some(_conn) = self.get_best_conn(&peer_id) {
            let _ = reply.send(Ok(()));
        } else {
            // dialing peer, with the post-processing callback
            self.dial_peer(peer_id, use_routing, |r: Result<&mut Connection>| {
                let _ = reply.send(r.map(|_| ()));
            });
        }
        Ok(())
    }

    fn on_close_connection(&mut self, peer_id: PeerId, reply: oneshot::Sender<Result<()>>) -> Result<()> {
        if let Some(ids) = self.connections_by_peer.get(&peer_id) {
            // close all connections related to the peer_id
            for id in ids {
                if let Some(c) = self.connections_by_id.get(&id) {
                    c.close()
                }
            }
        }

        let _ = reply.send(Ok(()));
        Ok(())
    }

    fn on_new_stream(
        &mut self,
        peer_id: PeerId,
        pids: Vec<ProtocolId>,
        use_routing: bool,
        reply: oneshot::Sender<Result<Substream>>,
    ) -> Result<()> {
        if let Some(connection) = self.get_best_conn(&peer_id) {
            // well, we have a connection, start a task to open the stream
            log::debug!(
                "open a stream using the existing connection {:?} for {:?}",
                connection.id(),
                peer_id
            );
            connection.open_stream(pids, move |r| {
                let _ = reply.send(r.map_err(|e| e.into()));
            });
        } else {
            log::debug!("dial and open a stream for {:?}", peer_id);
            // dialing peer, and opening a new stream in the post-processing callback
            self.dial_peer(peer_id.clone(), use_routing, |r: Result<&mut Connection>| match r {
                Ok(connection) => {
                    connection.open_stream(pids, |r| {
                        let _ = reply.send(r.map_err(|e| e.into()));
                    });
                }
                Err(e) => {
                    let _ = reply.send(Err(e));
                }
            });
        }
        Ok(())
    }

    fn on_close_stream(&mut self, cid: ConnectionId, sid: StreamId) -> Result<()> {
        self.handle_stream_closed(cid, sid)
    }

    ///
    fn on_retrieve_own_addresses(&mut self, f: impl FnOnce(Vec<Multiaddr>)) -> Result<()> {
        f(self.get_self_addrs());
        Ok(())
    }

    /// Retrieves the network information.
    fn on_retrieve_network_info(&mut self, f: impl FnOnce(NetworkInfo)) -> Result<()> {
        f(self.get_network_info());
        Ok(())
    }
    /// Retrieves the indentify information.
    fn on_retrieve_identify_info(&mut self, f: impl FnOnce(IdentifyInfo)) -> Result<()> {
        f(self.get_identify_info());
        Ok(())
    }
    /// Retrieves the connection views.
    fn on_retrieve_connection_views(&mut self, pid: Option<PeerId>, f: impl FnOnce(Vec<ConnectionView>)) -> Result<()> {
        f(self.get_connection_views(pid));
        Ok(())
    }
    /// Retrieves the substream views.
    fn on_retrieve_substream_views(&mut self, pid: PeerId, f: impl FnOnce(Result<Vec<SubstreamView>>)) -> Result<()> {
        f(self.get_substream_views(pid));
        Ok(())
    }
    /// Starts Swarm background task
    /// handling the internal events and external controls
    pub fn start(self) {
        // well, self 'move' explicitly,
        let mut swarm = self;

        // start GC task for peer store
        let peer_store = swarm.peer_store.clone();
        let (mut tx, mut rx) = mpsc::channel::<()>(0);
        // The GC task is to remove all expired addresses from the peer store
        task::spawn(async move {
            log::info!("starting Peerstore GC...");
            loop {
                let either = future::select(rx.next(), task::sleep(PEERSTORE_GC_PURGE_INTERVAL).boxed()).await;
                match either {
                    Either::Left((_, _)) => break,
                    Either::Right((_, _)) => peer_store.remove_expired_addrs(),
                }
            }
            log::info!("quitting Peerstore GC...");
        });
        task::spawn(async move {
            log::info!("starting Swarm main loop...");
            let r = swarm.handle_messages().await;
            log::info!("quitting Swarm main loop, due to {:?}...", r);

            // close peerstore GC task
            tx.close_channel();

            if let Err(e) = swarm.peer_store.save_data() {
                log::info!("PeerStore save data failed: {}", e);
            }
            log::info!("Swarm main loop exited");
        });
    }

    fn get_connection_views(&self, peer_id: Option<PeerId>) -> Vec<ConnectionView> {
        self.connections_by_id
            .values()
            .filter(|c| peer_id.as_ref().map_or(true, |pid| pid == &c.remote_peer()))
            .map(|c| c.to_view())
            .collect()
    }
    fn get_substream_views(&self, peer_id: PeerId) -> Result<Vec<SubstreamView>> {
        let r = self.connections_by_peer.get(&peer_id);

        if let Some(cids) = r {
            let c = cids
                .iter()
                .filter_map(|cid| self.connections_by_id.get(cid).map(|c| c.substream_view()))
                .flatten()
                .collect();
            Ok(c)
        } else {
            Err(SwarmError::NoConnection(peer_id))
        }
    }
    /// Returns network information about the `Swarm`.
    fn get_network_info(&self) -> NetworkInfo {
        let id = self.local_peer_id().clone();
        let num_connections = self.connections_by_id.len();
        let num_peers = self.connections_by_peer.len();
        let num_active_streams = self.connections_by_id.iter().fold(0, |acc, (_k, v)| acc + v.num_streams());
        // TODO complete this struct
        NetworkInfo {
            id,
            num_peers,
            num_connections,
            num_connections_pending: 0,
            num_connections_established: 0,
            num_active_streams,
            connection_info: vec![],
        }
    }
    /// Returns identify information about the `Swarm`.
    fn get_identify_info(&self) -> IdentifyInfo {
        let protocols = self.muxer.supported_protocols().into_iter().map(|p| p.to_string()).collect();

        let public_key = self.public_key.clone();
        let listen_addrs = self.get_self_addrs();

        IdentifyInfo {
            public_key,
            protocol_version: LIBP2P_RS_PROTOCOL_VERSION.to_string(),
            agent_version: LIBP2P_RS_AGENT_VERSION.to_string(),
            listen_addrs,
            protocols,
        }
    }

    /// Starts listening on the given address.
    ///
    /// Returns an error if the address is not supported.
    pub fn listen_on(&mut self, addrs: Vec<Multiaddr>) -> Result<()> {
        let mut succeeded: u32 = 0;
        let mut errs = FnvHashMap::<u32, SwarmError>::default();
        for (i, n) in addrs.clone().into_iter().enumerate() {
            let r = self.add_listen_addr(n);
            match r {
                Ok(()) => succeeded += 1,
                Err(e) => {
                    errs.insert(i as u32, e);
                }
            };
        }

        for (i, err) in errs.into_iter().enumerate() {
            log::warn!("listen on {} failed: {:?}", addrs[i], err)
        }

        if succeeded == 0 && !addrs.is_empty() {
            return Err(SwarmError::CanNotListenOnAny);
        }
        Ok(())
    }

    fn get_self_addrs(&self) -> Vec<Multiaddr> {
        // build self addrs with the self.listened_addrs + self.external_addrs
        let mut listen_addrs = self.external_addrs.iter().cloned().collect::<Vec<_>>();
        listen_addrs.extend(self.listened_addrs.to_vec());

        listen_addrs.dedup();

        log::trace!("swarm self addresses: {:?}", listen_addrs);

        listen_addrs
    }

    fn add_listen_addr(&mut self, addr: Multiaddr) -> Result<()> {
        log::info!("starting a listener on {:?}", addr);
        let mut transport = self.transports.lookup_by_addr(addr.clone())?;
        let mut listener = transport.listen_on(addr)?;
        if let Some(addr) = listener.multi_addr() {
            log::info!("adding an actual listening address {}", addr);
            self.listened_addrs.push(addr.clone());
        }

        let mut tx = self.event_sender.clone();
        // start a task for this listener
        // TODO: remember the task handle of this listener, so that we can 'cancel' it when exiting
        task::spawn(async move {
            loop {
                let r = listener.accept().await;
                match r {
                    Ok(ListenerEvent::AddressAdded(addr)) => {
                        let _ = tx.send(SwarmEvent::ListenAddressAdded(addr)).await;
                    }
                    Ok(ListenerEvent::AddressDeleted(addr)) => {
                        let _ = tx.send(SwarmEvent::ListenAddressDeleted(addr)).await;
                    }
                    Ok(ListenerEvent::Accepted(muxer)) => {
                        // don't have to verify if remote peer id matches its public key
                        // always accept any incoming connection
                        // send muxer back to Swarm main task
                        let _ = tx
                            .send(SwarmEvent::ConnectionEstablished {
                                stream_muxer: muxer,
                                direction: Direction::Inbound,
                                tid: None,
                            })
                            .await;
                    }
                    Err(err) => {
                        let _ = tx
                            .send(SwarmEvent::IncomingConnectionError {
                                remote_addr: Multiaddr::empty(),
                                error: err,
                            })
                            .await;
                    }
                }
            }
        });
        Ok(())
    }

    //
    fn dial_addr<F: FnOnce(Result<&mut Connection>) + Send + 'static>(&mut self, peer_id: PeerId, addrs: Vec<Multiaddr>, f: F) {
        log::debug!("dialing {:?} with addrs={:?}", peer_id, addrs);
        // if dialing to itself...
        if self.local_peer_id().eq(&peer_id) {
            f(Err(SwarmError::DialToSelf));
            return;
        }

        // allocate transaction id and push box::f into hashmap for post-processing
        let tid = self.assign_tid();
        self.dial_transactions.insert(tid, Box::new(f));
        self.dialer.dial(
            peer_id,
            self.transports.clone(),
            EitherDialAddr::Addresses(addrs),
            self.event_sender.clone(),
            tid,
        );
    }

    // dial the peer. try to get the multiaddr from peer store, otherwise use the
    // routing interface to find out if routing is available
    fn dial_peer<F: FnOnce(Result<&mut Connection>) + Send + 'static>(&mut self, peer_id: PeerId, use_routing: bool, f: F) {
        log::debug!("dialing {:?}, with routing={}", peer_id, use_routing);
        // if dialing to itself...
        if self.local_peer_id().eq(&peer_id) {
            f(Err(SwarmError::DialToSelf));
            return;
        }
        // then check addrs, return error if None while routing is not available
        let addrs = match self.peer_store.get_addrs(&peer_id) {
            Some(list) if !list.is_empty() => dial::EitherDialAddr::Addresses(list),
            _ => {
                if use_routing && self.routing.is_some() {
                    // ok, clone the routing interface into EitherDialAddr
                    let routing = self.routing.as_ref().unwrap().box_clone();
                    dial::EitherDialAddr::DHT(routing)
                } else {
                    // if routing is NOT available, reply with NoAddresses
                    f(Err(SwarmError::NoAddresses(peer_id)));
                    return;
                }
            }
        };

        // allocate transaction id and push box::f into hashmap for post-processing
        let tid = self.assign_tid();
        self.dial_transactions.insert(tid, Box::new(f));
        self.dialer
            .dial(peer_id, self.transports.clone(), addrs, self.event_sender.clone(), tid);
    }

    fn get_best_conn(&mut self, peer_id: &PeerId) -> Option<&mut connection::Connection> {
        let mut best = None;
        // selects the best connection we have to the peer.
        // TODO: we might have multiple connections towards a PeerId

        log::trace!("trying to get the best connnection for {:?}", peer_id);

        // self.connections_by_peer
        //     .iter()
        //     .for_each(|(k, v)| log::trace!("get best conn,{:?}={:?}", k, v));

        //let v = self.connections_by_peer.get_mut(peer_id).unwrap();

        if let Some(ids) = self.connections_by_peer.get(peer_id) {
            // TODO: to check if this connection is being closed

            let mut len = 0;
            for id in ids.iter() {
                if let Some(connection) = self.connections_by_id.get(id) {
                    let num = connection.num_streams();
                    if num >= len {
                        len = num;
                        best = Some(id);
                    }
                }
            }
        }

        if let Some(id) = best {
            self.connections_by_id.get_mut(id)
        } else {
            None
        }
    }

    fn is_connected(&self, peer_id: &PeerId) -> bool {
        // TODO: check if the connection is being closed??
        self.connections_by_peer.get(peer_id).map_or(0, |v| v.len()) > 0
    }

    /// Returns the peer ID of the swarm passed as parameter.
    pub fn local_pubkey(&self) -> &PublicKey {
        &self.public_key
    }

    /// Returns the peer ID of the swarm passed as parameter.
    pub fn local_peer_id(&self) -> &PeerId {
        &self.local_peer_id
    }

    /// Bans a peer by its peer ID.
    ///
    /// Any incoming connection and any dialing attempt will immediately be rejected.
    /// This function has no effect is the peer is already banned.
    pub fn ban_peer_id(&mut self, peer_id: PeerId) {
        self.banned_peers.insert(peer_id);

        // TODO: to disconnect
        // if let Some(c) = self.network.peer(peer_id).into_connected() {
        //     c.disconnect();
        // }
    }

    /// Unbans a peer.
    pub fn unban_peer_id(&mut self, peer_id: PeerId) {
        self.banned_peers.remove(peer_id.as_ref());
    }

    fn add_connection(&mut self, connection: Connection) {
        let cid = connection.id();
        let remote_peer_id = connection.remote_peer();

        // append to the by_id hashmap
        self.connections_by_id.insert(cid, connection);

        // append to the by peer hashmap
        let conns = self.connections_by_peer.entry(remote_peer_id).or_default();
        conns.push(cid);

        log::trace!("connection added to hashmap, total={}", self.connections_by_id.len());

        // TODO: we have a connection to the specified peer_id, now cancel all pending attempts

        // TODO: generate a connected event

        // TODO: return the connection
    }

    /// Handles a new connection.
    ///
    /// start a Task for accepting new sub-stream from the connection
    fn handle_connection_opened(&mut self, stream_muxer: IStreamMuxer, dir: Direction, tid: Option<TransactionId>) -> Result<()> {
        log::debug!("handle_connection_opened: {:?} {:?}", stream_muxer, dir);

        let metric = self.metric.clone();

        // add local pubkey to keybook
        // self.peer_store.add_key(&self.local_peer_id, stream_muxer.local_priv_key().public());

        // clone the stream_muxer, and then wrap into Connection, task_handle will be assigned later
        let mut connection = Connection::new(
            self.assign_cid(),
            stream_muxer.clone(),
            dir,
            self.event_sender.clone(),
            self.ctrl_sender.clone(),
            metric.clone(),
        );
        // TODO: filtering the multiaddr, Err = AddrFiltered(addr)

        /*        raddr := tc.RemoteMultiaddr()
                if s.Filters.AddrBlocked(raddr) {
                    tc.Close()
                    return nil, ErrAddrFiltered
                }

                p := tc.RemotePeer()

                // Add the public key.
                if pk := tc.RemotePublicKey(); pk != nil {
                    s.peers.AddPubKey(p, pk)
                }
        */
        // TODO: add remote pubkey to keystore

        let mut tx = self.event_sender.clone();
        let cid = connection.id();
        // clone muxer and move it into the task
        let mut muxer = self.muxer.clone();
        let ctrl = self.ctrl_sender.clone();

        // Note we have to use the original copy of the stream muxer to start the task,
        // instead of the cloned one which doesn't have the task handle at all
        let handle = task::spawn(async move {
            let mut stream_muxer = stream_muxer;
            // start the background task of the stream_muxer, the handle can be await'ed by us
            let task_handle = stream_muxer.task().map(task::spawn);
            loop {
                let metric = metric.clone();
                let ctrl = ctrl.clone();
                let r = stream_muxer.accept_stream().await;

                // TODO: probably we should spawn a new task for protocol selection
                // As a fact, in go-libp2p, the protocol selection is done in a blocking way...
                match r {
                    Ok(raw_stream) => {
                        // well, got the raw stream, now do protocol selection
                        log::debug!("run protocol selection for inbound stream={:?}", raw_stream);

                        // now it's time to do multistream multiplexing for inbound stream
                        let result = muxer.select_inbound(raw_stream).await;
                        match result {
                            Ok((mut handler, raw_stream, proto)) => {
                                let la = stream_muxer.local_multiaddr();
                                let ra = stream_muxer.remote_multiaddr();
                                let rpid = stream_muxer.remote_peer();
                                let ci = ConnectInfo { la, ra, rpid };
                                let stream = Substream::new(raw_stream, metric, Direction::Inbound, proto.clone(), cid, ci, ctrl);
                                let view = stream.to_view();
                                let _ = tx.send(SwarmEvent::StreamOpened { view }).await;

                                // anyway, start handler task
                                task::spawn(async move {
                                    let _ = handler.handle(stream, proto).await;
                                });

                                // TODO: hook the task handle to the Substream, so that it can wait for exiting the task
                            }
                            Err(error) => {
                                log::debug!("failed inbound protocol selection {:?} {:?}", cid, error);
                                let _ = tx.send(SwarmEvent::StreamError { cid, error }).await;
                            }
                        }
                    }
                    Err(error) => {
                        log::debug!("connection closed {:?} {:?}", cid, error);
                        let _ = tx.send(SwarmEvent::ConnectionClosed { cid, error }).await;
                        // something happened, break the loop then exit the Task
                        break;
                    }
                }
            }

            // As stream_muxer is closed, we wait for its task_handle
            if let Some(h) = task_handle {
                h.await;
            }

            log::debug!("{:?} accept-task exiting...", stream_muxer);
        });

        // now we have the handle, move it into Connection
        connection.set_handle(handle);

        for handler in self.muxer.protocol_handlers.values_mut() {
            handler.connected(&mut connection);
        }

        // dial callback for post-processing
        // note that it must cleanup the tid entry
        if let Some(id) = tid {
            // the entry must be there
            log::debug!("invoking dial transaction {:?}", tid);
            let callback = self.dial_transactions.remove(&id).expect("no match tid found");
            callback(Ok(&mut connection));
        }

        // insert to the hashmap of connections
        // there might be a race condition:
        // the spawned connection task might have exited for some reason, before we insert connection
        // into the hashmap. No problem, the connection will be cleaned up in 'handle_connection_closed'
        // event.
        self.add_connection(connection);

        Ok(())
    }

    /// Handles outgoing connection error.
    fn handle_connection_error(&mut self, peer_id: PeerId, error: SwarmError, tid: TransactionId) -> Result<()> {
        log::debug!("handle_connection_error: {:?} {:?} tid={:?}", peer_id, error, tid);

        //execute dial callback for post processing
        let callback = self.dial_transactions.remove(&tid).expect("no match tid found");
        callback(Err(error));

        // TODO: add statistics
        Ok(())
    }

    /// Handles opening stream
    ///
    /// Use channel to received message that sent by another task
    fn handle_stream_opened(&mut self, view: SubstreamView) -> Result<()> {
        log::debug!("handle_stream_opened: {:?}", view);
        // add stream id to the connection substream list
        if let Some(c) = self.connections_by_id.get_mut(&view.cid) {
            c.add_stream(view)
        };
        Ok(())
    }

    /// Handles closing stream
    fn handle_stream_closed(&mut self, cid: ConnectionId, sid: StreamId) -> Result<()> {
        log::debug!("handle_stream_closed: {:?}/{:?}", cid, sid);
        // delete sub-stream from the connection substream list
        if let Some(c) = self.connections_by_id.get_mut(&cid) {
            c.del_stream(sid)
        }
        Ok(())
    }

    /// Handles closing a connection
    ///
    /// start a Task for accepting new sub-stream from the connection
    fn handle_connection_closed(&mut self, cid: ConnectionId) -> Result<()> {
        log::debug!("handle_connection_closed: {:?}", cid);

        // try to retrieve the Connection by looking up 'connections_by_id'
        if let Some(mut connection) = self.connections_by_id.remove(&cid) {
            for handler in self.muxer.protocol_handlers.values_mut() {
                handler.disconnected(&mut connection);
            }

            let remote_peer_id = connection.remote_peer();
            if let Some(ids) = self.connections_by_peer.get_mut(&remote_peer_id) {
                ids.retain(|id| id != &cid);
                // remove the peer if all the connections of the peer are closed
                if ids.is_empty() {
                    self.connections_by_peer.remove(&remote_peer_id);
                }
            } else {
                log::warn!("shouldn't happen, PeerId={:?}", remote_peer_id);
            }

            // now, close all tasks owned by the connection
            task::spawn(async move {
                let _ = connection.wait().await;
                let _ = connection.stop_ping().await;
                let _ = connection.stop_identify().await;
                let _ = connection.stop_identify_push().await;
            });
        } else {
            log::info!("shouldn't happen, wired connection {:?}", cid);
        }

        Ok(())
    }

    /// Received ping result, and then updates addrbook in peerstore
    fn handle_ping_result(&mut self, cid: ConnectionId, result: Result<Duration>) -> Result<()> {
        log::trace!("handle_ping_result: {:?} {:?}", cid, result);

        if let Some(connection) = self.connections_by_id.get_mut(&cid) {
            match result {
                Ok(ttl) => {
                    //let remote_peer_id = c.remote_peer();
                    log::trace!("ping TTL={:?} for {:?}", ttl, connection);
                    // update peer store with the TTL
                    let peer_id = connection.stream_muxer().remote_peer();
                    self.peer_store.update_addr(&peer_id, ttl);
                }
                Err(_) => {
                    log::info!("reach the max ping failure count, closing {:?}", connection);
                    connection.close();
                }
            }
        }

        Ok(())
    }

    fn handle_observed_address(&mut self, observed_addr: Multiaddr, cid: ConnectionId) {
        log::debug!("identify observed_addr: {} cid={:?}", observed_addr, cid);
        let addrs = self.address_translation(&observed_addr).collect::<Vec<_>>();
        for addr in addrs {
            self.external_addrs.add(addr);
        }

        log::debug!("external address: {:?}", self.external_addrs)
    }

    /// Received result which contains IdentityInfo and multiaddr,
    /// and then updates keybook and protobook in peerstore.
    fn handle_identify_result(&mut self, cid: ConnectionId, result: Result<(IdentifyInfo, Multiaddr)>) -> Result<()> {
        log::debug!("handle_identify_result: {:?}", cid);

        if let Some(connection) = self.connections_by_id.get_mut(&cid) {
            match result {
                Ok((info, observed_addr)) => {
                    let remote_pubkey = connection.remote_pub_key();
                    let peer_id = connection.remote_peer();

                    self.handle_observed_address(observed_addr, cid);

                    log::debug!(
                        "identified peer addresses {:?} protocols {:?} for {}",
                        info.listen_addrs,
                        info.protocols,
                        peer_id
                    );
                    // update peerstore with the listening addresses and protocols of the remote peer
                    // Note, we don't use connection.remote_addr(), because it might be a NATed address/port which
                    // changed very frequently. Instead, using info.listen_addrs is a better solution.
                    // TODO: to handle info.protocol_version .agent_version
                    self.peer_store.add_addrs(&peer_id, info.listen_addrs, ADDRESS_TTL);
                    self.peer_store.add_key(&peer_id, remote_pubkey);
                    self.peer_store.add_protocols(&peer_id, info.protocols);

                    // well, kick off all protocol handlers for the Identify completion
                    for handler in self.muxer.protocol_handlers.values_mut() {
                        handler.identified(peer_id.clone());
                    }
                }
                Err(err) => {
                    log::info!("identify failed {:?} for {:?}", err, connection);
                }
            }
        }

        Ok(())
    }

    /// Call this function in order to know which address remotes should dial to
    /// access your local node.
    ///
    /// When receiving an observed address on a tcp connection that we initiated, the observed
    /// address contains our tcp dial port, not our tcp listen port. We know which port we are
    /// listening on, thereby we can replace the port within the observed address.
    ///
    /// When receiving an observed address on a tcp connection that we did **not** initiated, the
    /// observed address should contain our listening port. In case it differs from our listening
    /// port there might be a proxy along the path.
    ///
    /// # Arguments
    ///
    /// * `observed_addr` - should be an address a remote observes you as, which can be obtained for
    /// example with the identify protocol.
    ///
    fn address_translation<'a>(&'a self, observed_addr: &'a Multiaddr) -> impl Iterator<Item = Multiaddr> + 'a {
        let mut addrs: Vec<_> = self
            .listened_addrs
            .iter()
            .filter_map(move |server| address_translation(server, observed_addr))
            .collect();

        // remove duplicates
        addrs.sort_unstable();
        addrs.dedup();

        addrs.into_iter()
    }
}

/// The possible failures of [`Swarm`].
#[derive(Debug)]
pub enum SwarmError {
    /// The configured limit for simultaneous outgoing connections
    /// has been reached.
    ConnectionLimit(ConnectionLimit),

    /// Returned no addresses for the peer to dial.
    NoAddresses(PeerId),

    /// No connection yet, unable to open a sub stream.
    NoConnection(PeerId),

    /// The peer identity obtained on the connection did not
    /// match the one that was expected.
    InvalidPeerId(PeerId),

    /// Closing down. Swarm is being closed at this moment
    /// 1: event channel, 2: ctrl channel
    Closing(u32),

    /// Transport Error
    ///
    /// Contains a TransportError.
    Transport(TransportError),

    /// Internal, tentatively for convenience
    Internal,

    /// listen on fail
    CanNotListenOnAny,

    ///ErrDialToSelf is returned if we attempt to dial our own peer
    DialToSelf,

    ///been dialed too frequently
    DialBackoff,

    ///AllDialsFailed is returned when connecting to a peer has ultimately failed
    AllDialsFailed,

    ///dial timeout
    DialTimeout(Multiaddr, u64),

    ///max dial attempts exceeded
    MaxDialAttempts(u32),

    ///max concurrent dial  exceeded
    ConcurrentDialLimit(u32),
}

#[rustfmt::skip]
impl fmt::Display for SwarmError {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        match self {
            SwarmError::ConnectionLimit(err) => write!(f, "Swarm Dial error: {}", err),
            SwarmError::NoAddresses(peer_id) => write!(f, "Swarm Dial error: no addresses for peer{:?}.", peer_id),
            SwarmError::NoConnection(peer_id) => write!(f, "Swarm Stream error: no connections for peer{:?}.", peer_id),
            SwarmError::InvalidPeerId(peer_id) => write!(f, "Swarm Dial error: invalid peer id{:?}.", peer_id),
            SwarmError::Transport(err) => write!(f, "Swarm Transport error: {}.", err),
            SwarmError::Internal => write!(f, "Swarm internal error."),
            SwarmError::Closing(s) => write!(f, "Swarm channel closed source={}.", s),
            SwarmError::CanNotListenOnAny => write!(f, "Failed to listen on any addresses"),
            SwarmError::DialToSelf => write!(f, "Swarm Dial error:dial to self attempted"),
            SwarmError::DialBackoff => write!(f, "Swarm Dial error:dial backoff"),
            SwarmError::AllDialsFailed => write!(f, "Swarm Dial error:all dials failed"),
            SwarmError::DialTimeout(ma, t) => write!(f, "Swarm Dial error:dial timeout, addr={:?},timeout={:?}", ma, Duration::from_secs(*t)),
            SwarmError::MaxDialAttempts(c) => write!(f, "Swarm Dial error:max dial attempts exceeded, count={}", c),
            SwarmError::ConcurrentDialLimit(c) => write!(f, "Swarm Dial error:max concurrent dial exceeded, count={}", c),
        }
    }
}

impl error::Error for SwarmError {
    fn source(&self) -> Option<&(dyn error::Error + 'static)> {
        match self {
            SwarmError::ConnectionLimit(err) => Some(err),
            SwarmError::NoAddresses(_) => None,
            SwarmError::NoConnection(_) => None,
            SwarmError::InvalidPeerId(_) => None,
            SwarmError::Transport(err) => Some(err),
            SwarmError::Internal => None,
            SwarmError::Closing(_) => None,
            SwarmError::CanNotListenOnAny => None,
            SwarmError::DialToSelf => None,
            SwarmError::DialBackoff => None,
            SwarmError::AllDialsFailed => None,
            SwarmError::DialTimeout(_, _) => None,
            SwarmError::MaxDialAttempts(_) => None,
            SwarmError::ConcurrentDialLimit(_) => None,
        }
    }
}

impl From<std::io::Error> for SwarmError {
    fn from(err: std::io::Error) -> Self {
        SwarmError::Transport(TransportError::IoError(err))
    }
}

impl From<TransportError> for SwarmError {
    fn from(err: TransportError) -> Self {
        SwarmError::Transport(err)
    }
}

impl From<mpsc::SendError> for SwarmError {
    // TODO: make a error catelog for SendError
    fn from(_: mpsc::SendError) -> Self {
        SwarmError::Internal
    }
}

impl From<oneshot::Canceled> for SwarmError {
    // TODO: make a error catelog for Canceled
    fn from(_: oneshot::Canceled) -> Self {
        SwarmError::Internal
    }
}