<commit_before>#![allow(dead_code)]

use arrayvec::ArrayString;
use async_trait::async_trait;
use devp2p::*;
use ethereum_types::*;
use futures::stream::BoxStream;
use hex_literal::hex;
use k256::ecdsa::SigningKey;
use maplit::btreemap;
use parking_lot::RwLock;
use rand::rngs::OsRng;
use rlp_derive::{RlpDecodable, RlpEncodable};
use std::{
    collections::{BTreeSet, HashMap},
    sync::{
        atomic::{AtomicUsize, Ordering},
        Arc,
    },
};
use task_group::TaskGroup;
use tokio::{
    stream::StreamExt,
    sync::{
        mpsc::{channel, Sender},
        Mutex as AsyncMutex,
    },
};
use tracing::*;
use tracing_subscriber::EnvFilter;
use trust_dns_resolver::{config::*, TokioAsyncResolver};
use uuid::Uuid;

const DNS_BOOTNODE: &str = "all.mainnet.ethdisco.net";

fn eth() -> CapabilityName {
    CapabilityName(ArrayString::from("eth").unwrap())
}

#[derive(Debug, Default)]
struct TaskMetrics {
    count: AtomicUsize,
}

impl task_group::Metrics for TaskMetrics {
    fn task_started(&self, id: Uuid, name: String) {
        let c = self.count.fetch_add(1, Ordering::Relaxed);
        trace!("Current tasks: {}. Started task {}/{}", c + 1, name, id)
    }

    fn task_stopped(&self, id: Uuid, name: String) {
        let c = self.count.fetch_sub(1, Ordering::Relaxed);
        trace!("Current tasks: {}. Stopped task {}/{}", c - 1, name, id)
    }
}

#[derive(Debug, RlpEncodable, RlpDecodable)]
struct StatusMessage {
    protocol_version: usize,
    network_id: usize,
    total_difficulty: U256,
    best_hash: H256,
    genesis_hash: H256,
}

#[derive(Clone)]
struct Pipes {
    sender: Sender<OutboundEvent>,
    receiver: Arc<AsyncMutex<BoxStream<'static, OutboundEvent>>>,
}

#[derive(Default)]
struct CapabilityServerImpl {
    on_event_pipes: Arc<RwLock<HashMap<PeerId, Pipes>>>,
}

impl CapabilityServerImpl {
    fn setup_pipes(&self, peer: PeerId, pipes: Pipes) {
        assert!(self.on_event_pipes.write().insert(peer, pipes).is_none());
    }
    fn get_pipes(&self, peer: PeerId) -> Pipes {
        self.on_event_pipes.read().get(&peer).unwrap().clone()
    }
    fn connected_peers(&self) -> usize {
        self.on_event_pipes.read().len()
    }
}

#[async_trait]
impl CapabilityServer for CapabilityServerImpl {
    #[instrument(skip(self))]
    fn on_peer_connect(&self, peer: PeerId, _: BTreeSet<CapabilityId>) {
        info!("Settting up peer state");
        let status_message = StatusMessage {
            protocol_version: 63,
            network_id: 1,
            total_difficulty: 17608636743620256866935_u128.into(),
            best_hash: H256::from(hex!(
                "28042e7e4d35a3482bf5f0d862501868b04c1734f483ceae3bf1393561951829"
            )),
            genesis_hash: H256::from(hex!(
                "d4e56740f876aef8c010b86a40d5f56745a118d0906a34e69aec8c0db1cb8fa3"
            )),
        };

        let (sender, receiver) = channel(1);
        let receiver = Box::pin(
            tokio::stream::iter(std::iter::once(OutboundEvent::Message {
                capability_name: eth(),
                message: Message {
                    id: 0,
                    data: rlp::encode(&status_message).into(),
                },
            }))
            .chain(receiver),
        );
        self.setup_pipes(
            peer,
            Pipes {
                sender,
                receiver: Arc::new(AsyncMutex::new(receiver)),
            },
        );
    }
    #[instrument(skip(self))]
    async fn on_peer_event(&self, peer: PeerId, event: InboundEvent) {
        match event {
            InboundEvent::Disconnect { .. } => {
                self.on_event_pipes.write().remove(&peer);
            }
            InboundEvent::Message { message, .. } => {
                info!(
                    "Received message with id {}, data {}",
                    message.id,
                    hex::encode(&message.data)<commit_msg>sentry: more concise code, teardown<commit_after>#![allow(dead_code)]

use arrayvec::ArrayString;
use async_trait::async_trait;
use devp2p::*;
use ethereum_types::*;
use futures::stream::BoxStream;
use hex_literal::hex;
use k256::ecdsa::SigningKey;
use maplit::btreemap;
use parking_lot::RwLock;
use rand::rngs::OsRng;
use rlp_derive::{RlpDecodable, RlpEncodable};
use std::{
    collections::{BTreeSet, HashMap},
    sync::{
        atomic::{AtomicUsize, Ordering},
        Arc,
    },
};
use task_group::TaskGroup;
use tokio::{
    stream::StreamExt,
    sync::{
        mpsc::{channel, Sender},
        Mutex as AsyncMutex,
    },
};
use tracing::*;
use tracing_subscriber::EnvFilter;
use trust_dns_resolver::{config::*, TokioAsyncResolver};
use uuid::Uuid;

const DNS_BOOTNODE: &str = "all.mainnet.ethdisco.net";

fn eth() -> CapabilityName {
    CapabilityName(ArrayString::from("eth").unwrap())
}

#[derive(Debug, Default)]
struct TaskMetrics {
    count: AtomicUsize,
}

impl task_group::Metrics for TaskMetrics {
    fn task_started(&self, id: Uuid, name: String) {
        let c = self.count.fetch_add(1, Ordering::Relaxed);
        trace!("Current tasks: {}. Started task {}/{}", c + 1, name, id)
    }

    fn task_stopped(&self, id: Uuid, name: String) {
        let c = self.count.fetch_sub(1, Ordering::Relaxed);
        trace!("Current tasks: {}. Stopped task {}/{}", c - 1, name, id)
    }
}

#[derive(Debug, RlpEncodable, RlpDecodable)]
struct StatusMessage {
    protocol_version: usize,
    network_id: usize,
    total_difficulty: U256,
    best_hash: H256,
    genesis_hash: H256,
}

#[derive(Clone)]
struct Pipes {
    sender: Sender<OutboundEvent>,
    receiver: Arc<AsyncMutex<BoxStream<'static, OutboundEvent>>>,
}

#[derive(Default)]
struct CapabilityServerImpl {
    peer_pipes: Arc<RwLock<HashMap<PeerId, Pipes>>>,
}

impl CapabilityServerImpl {
    fn setup_pipes(&self, peer: PeerId, pipes: Pipes) {
        assert!(self.peer_pipes.write().insert(peer, pipes).is_none());
    }
    fn get_pipes(&self, peer: PeerId) -> Pipes {
        self.peer_pipes.read().get(&peer).unwrap().clone()
    }
    fn teardown(&self, peer: PeerId) {
        self.peer_pipes.write().remove(&peer);
    }
    fn connected_peers(&self) -> usize {
        self.peer_pipes.read().len()
    }
}

#[async_trait]
impl CapabilityServer for CapabilityServerImpl {
    #[instrument(skip(self))]
    fn on_peer_connect(&self, peer: PeerId, _: BTreeSet<CapabilityId>) {
        info!("Settting up peer state");
        let status_message = StatusMessage {
            protocol_version: 63,
            network_id: 1,
            total_difficulty: 17608636743620256866935_u128.into(),
            best_hash: H256::from(hex!(
                "28042e7e4d35a3482bf5f0d862501868b04c1734f483ceae3bf1393561951829"
            )),
            genesis_hash: H256::from(hex!(
                "d4e56740f876aef8c010b86a40d5f56745a118d0906a34e69aec8c0db1cb8fa3"
            )),
        };

        let first_message = OutboundEvent::Message {
            capability_name: eth(),
            message: Message {
                id: 0,
                data: rlp::encode(&status_message).into(),
            },
        };

        let (sender, receiver) = channel(1);
        let receiver =
            Box::pin(tokio::stream::iter(std::iter::once(first_message)).chain(receiver));
        self.setup_pipes(
            peer,
            Pipes {
                sender,
                receiver: Arc::new(AsyncMutex::new(receiver)),
            },
        );
    }
    #[instrument(skip(self))]
    async fn on_peer_event(&self, peer: PeerId, event: InboundEvent) {
        match event {
            InboundEvent::Disconnect { .. } => {
                self.teardown(peer);
            }
            InboundEvent::Message { message, .. } => {
                info!(
                    "Received message with id {}, data {}",
                    message.id,
                    hex::encode(&message.data)
{'fraction_non_alphanumeric': 0.14028868067865283, 'fraction_numerical': 0.03545201316789061, 'mean_word_length': 2.679087098276665, 'pattern_counts': {'":': 0, '<': 19, '<?xml version=': 0, '>': 29, 'https://': 0, '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': '5586746', 'n_tokens_mistral': 2788, 'n_tokens_neox': 2659, 'n_words': 469}