<commit_before>use crate::behaviour::{Behaviour, BehaviourEvent, PubsubMessage};
use crate::error;
use crate::multiaddr::Protocol;
use crate::rpc::RPCEvent;
use crate::NetworkConfig;
use futures::prelude::*;
use futures::Stream;
use libp2p::core::{
    identity,
    muxing::StreamMuxerBox,
    nodes::Substream,
    transport::boxed::Boxed,
    upgrade::{InboundUpgradeExt, OutboundUpgradeExt},
};
use libp2p::identify::protocol::IdentifyInfo;
use libp2p::{core, secio, PeerId, Swarm, Transport};
use slog::{debug, info, trace, warn};
use std::io::{Error, ErrorKind};
use std::time::Duration;
use types::{TopicBuilder, TopicHash};

type Libp2pStream = Boxed<(PeerId, StreamMuxerBox), Error>;
type Libp2pBehaviour = Behaviour<Substream<StreamMuxerBox>>;

/// The configuration and state of the libp2p components for the beacon node.
pub struct Service {
    /// The libp2p Swarm handler.
    //TODO: Make this private
    pub swarm: Swarm<Libp2pStream, Libp2pBehaviour>,
    /// This node's PeerId.
    _local_peer_id: PeerId,
    /// The libp2p logger handle.
    pub log: slog::Logger,
}

impl Service {
    pub fn new(config: NetworkConfig, log: slog::Logger) -> error::Result<Self> {
        debug!(log, "Libp2p Service starting");

        // TODO: Currently using secp256k1 key pairs. Wire protocol specifies RSA. Waiting for this
        // PR to be merged to generate RSA keys: https://github.com/briansmith/ring/pull/733
        // TODO: Save and recover node key from disk
        let local_private_key = identity::Keypair::generate_secp256k1();

        let local_public_key = local_private_key.public();
        let local_peer_id = PeerId::from(local_private_key.public());
        info!(log, "Local peer id: {:?}", local_peer_id);

        let mut swarm = {
            // Set up the transport
            let transport = build_transport(local_private_key);
            // Set up gossipsub routing
            let behaviour = Behaviour::new(local_public_key.clone(), &config, &log);
            // Set up Topology
            let topology = local_peer_id.clone();
            Swarm::new(transport, behaviour, topology)
        };

        // listen on all addresses
        for address in config.listen_addresses().expect("invalid listen multiaddr") {
            match Swarm::listen_on(&mut swarm, address.clone()) {
                Ok(mut listen_addr) => {
                    listen_addr.append(Protocol::P2p(local_peer_id.clone().into()));
                    info!(log, "Listening on: {}", listen_addr);
                }
                Err(err) => warn!(log, "Cannot listen on: {} : {:?}", address, err),
            };
        }
        // connect to boot nodes - these are currently stored as multiaddrs
        // Once we have discovery, can set to peerId
        for bootnode in config.boot_nodes().expect("invalid boot node multiaddr") {
            match Swarm::dial_addr(&mut swarm, bootnode.clone()) {
                Ok(()) => debug!(log, "Dialing bootnode: {}", bootnode),
                Err(err) => debug!(
                    log,
                    "Could not connect to bootnode: {} error: {:?}", bootnode, err
                ),
            };
        }

        // subscribe to default gossipsub topics
        let mut subscribed_topics = vec![];
        for topic in config.topics {
            let t = TopicBuilder::new(topic.to_string()).build();
            if swarm.subscribe(t) {
                trace!(log, "Subscribed to topic: {:?}", topic);
                subscribed_topics.push(topic);
            } else {
                warn!(log, "Could not subscribe to topic: {:?}", topic)
            }
        }
        info!(log, "Subscribed to topics: {:?}", subscribed_topics);

        Ok(Service {
            _local_peer_id: local_peer_id,
            swarm,
            log,
        })
    }
}
<commit_msg>Return errors instead of panic in libp2p<commit_after>use crate::behaviour::{Behaviour, BehaviourEvent, PubsubMessage};
use crate::error;
use crate::multiaddr::Protocol;
use crate::rpc::RPCEvent;
use crate::NetworkConfig;
use futures::prelude::*;
use futures::Stream;
use libp2p::core::{
    identity,
    muxing::StreamMuxerBox,
    nodes::Substream,
    transport::boxed::Boxed,
    upgrade::{InboundUpgradeExt, OutboundUpgradeExt},
};
use libp2p::identify::protocol::IdentifyInfo;
use libp2p::{core, secio, PeerId, Swarm, Transport};
use slog::{debug, info, trace, warn};
use std::io::{Error, ErrorKind};
use std::time::Duration;
use types::{TopicBuilder, TopicHash};

type Libp2pStream = Boxed<(PeerId, StreamMuxerBox), Error>;
type Libp2pBehaviour = Behaviour<Substream<StreamMuxerBox>>;

/// The configuration and state of the libp2p components for the beacon node.
pub struct Service {
    /// The libp2p Swarm handler.
    //TODO: Make this private
    pub swarm: Swarm<Libp2pStream, Libp2pBehaviour>,
    /// This node's PeerId.
    _local_peer_id: PeerId,
    /// The libp2p logger handle.
    pub log: slog::Logger,
}

impl Service {
    pub fn new(config: NetworkConfig, log: slog::Logger) -> error::Result<Self> {
        debug!(log, "Libp2p Service starting");

        // TODO: Currently using secp256k1 key pairs. Wire protocol specifies RSA. Waiting for this
        // PR to be merged to generate RSA keys: https://github.com/briansmith/ring/pull/733
        // TODO: Save and recover node key from disk
        let local_private_key = identity::Keypair::generate_secp256k1();

        let local_public_key = local_private_key.public();
        let local_peer_id = PeerId::from(local_private_key.public());
        info!(log, "Local peer id: {:?}", local_peer_id);

        let mut swarm = {
            // Set up the transport
            let transport = build_transport(local_private_key);
            // Set up gossipsub routing
            let behaviour = Behaviour::new(local_public_key.clone(), &config, &log);
            // Set up Topology
            let topology = local_peer_id.clone();
            Swarm::new(transport, behaviour, topology)
        };

        // listen on all addresses
        for address in config
            .listen_addresses()
            .map_err(|e| format!("Invalid listen multiaddr: {}", e))?
        {
            match Swarm::listen_on(&mut swarm, address.clone()) {
                Ok(mut listen_addr) => {
                    listen_addr.append(Protocol::P2p(local_peer_id.clone().into()));
                    info!(log, "Listening on: {}", listen_addr);
                }
                Err(err) => warn!(log, "Cannot listen on: {} : {:?}", address, err),
            };
        }
        // connect to boot nodes - these are currently stored as multiaddrs
        // Once we have discovery, can set to peerId
        for bootnode in config
            .boot_nodes()
            .map_err(|e| format!("Invalid boot node multiaddr: {:?}", e))?
        {
            match Swarm::dial_addr(&mut swarm, bootnode.clone()) {
                Ok(()) => debug!(log, "Dialing bootnode: {}", bootnode),
                Err(err) => debug!(
                    log,
                    "Could not connect to bootnode: {} error: {:?}", bootnode, err
                ),
            };
        }

        // subscribe to default gossipsub topics
        let mut subscribed_topics = vec![];
        for topic in config.topics {
            let t = TopicBuilder::new(topic.to_string()).build();
            if swarm.subscribe(t) {
                trace!(log, "Subscribed to topic: {:?}", topic);
                subscribed_topics.push(topic);
            } else {
                warn!(log, "Could not subscribe to topic: {:?}", topic)
            }
        }
        info!(log, "Subscribed to topics: {:?}", subscribed_topics);

        Ok(Service {
            _local_peer_id: local_peer_id,
            swarm,
            log,
        })
    }
}

{'fraction_non_alphanumeric': 0.12740476493820868, 'fraction_numerical': 0.005988023952095809, 'mean_word_length': 2.364766395199314, 'pattern_counts': {'":': 0, '<': 13, '<?xml version=': 0, '>': 23, 'https://': 2, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1198205', 'n_tokens_mistral': 2363, 'n_tokens_neox': 2246, 'n_words': 600}