<commit_before>//! Ethereum DPT protocol implementation

extern crate etcommon_bigint as bigint;
extern crate etcommon_crypto as hash;
extern crate etcommon_rlp as rlp;
extern crate sha3;
extern crate secp256k1;
#[macro_use]
extern crate futures;
extern crate tokio_io;
extern crate tokio_core;
extern crate time;

mod proto;
mod message;
mod util;

use message::*;
use proto::{DPTCodec, DPTCodecMessage};
use futures::future;
use futures::{Poll, Async, StartSend, AsyncSink, Future, Stream, Sink};
use tokio_io::{AsyncRead, AsyncWrite};
use tokio_io::codec::{Framed, Encoder, Decoder};
use tokio_core::reactor::{Timeout, Handle};
use tokio_core::net::{UdpSocket, UdpFramed};
use std::net::{IpAddr, SocketAddr, Ipv4Addr, Ipv6Addr};
use std::io;
use bigint::{H256, H512};
use rlp::UntrustedRlp;
use hash::SECP256K1;
use secp256k1::key::{PublicKey, SecretKey};
use util::pk2id;

fn retain_mut<T, F>(vec: &mut Vec<T>, mut f: F)
    where F: FnMut(&mut T) -> bool
{
    let len = vec.len();
    let mut del = 0;
    {
        let v = &mut **vec;

        for i in 0..len {
            if !f(&mut v[i]) {
                del += 1;
            } else if del > 0 {
                v.swap(i - del, i);
            }
        }
    }
    if del > 0 {
        vec.truncate(len - del);
    }
}

/// DPT message for requesting new peers or ping with timeout
pub enum DPTMessage {
    RequestNewPeer,
    Ping(Timeout),
}

/// DPT stream for sending DPT messages or receiving new peers
pub struct DPTStream {
    stream: UdpFramed<DPTCodec>,
    id: H512,
    connected: Vec<DPTNode>,
    timeout: Option<(Timeout, Vec<H512>)>,
    incoming: Vec<DPTNode>,
    address: IpAddr,
    udp_port: u16,
    tcp_port: u16,
}

#[derive(Debug, Clone)]
/// DPT node used by a DPT stream
pub struct DPTNode {
    pub address: IpAddr,
    pub tcp_port: u16,
    pub udp_port: u16,
    pub id: H512,
}

impl DPTNode {
    /// The TCP socket address of this node
    pub fn tcp_addr(&self) -> SocketAddr {
        SocketAddr::new(self.address, self.tcp_port)
    }

    /// The UDP socket address of this node
    pub fn udp_addr(&self) -> SocketAddr {
        SocketAddr::new(self.address, self.udp_port)
    }
}

impl DPTStream {
    /// Create a new DPT stream
    pub fn new(addr: &SocketAddr, handle: &Handle,
               secret_key: SecretKey,
               bootstrap_nodes: Vec<DPTNode>,
               tcp_port: u16) -> Result<Self, io::Error> {
        let id = pk2id(&match PublicKey::from_secret_key(&SECP256K1, &secret_key) {
            Ok(val) => val,
            Err(_) => return Err(io::Error::new(io::ErrorKind::Other, "converting pub key failed")),
        });
        Ok(Self {
            stream: UdpSocket::bind(addr, handle)?.framed(DPTCodec::new(secret_key)),
            id, connected: bootstrap_nodes, incoming: Vec::new(),
            timeout: None,
            address: addr.ip(), udp_port: addr.port(), tcp_port
        })
    }
}

impl Stream for DPTStream {
    type Item = DPTNode;
    type Error = io::Error;

    fn poll(&mut self) -> Poll<Option<Self::Item>, Self::Error> {
        let mut timeoutted = false;
        if self.timeout.is_some() {
            let &mut (ref mut timeout, ref hs) = self.timeout.as_mut().unwrap();
            timeoutted = match timeout.poll() {
                Ok(Async::Ready(())) => true,
                Ok(Async::NotReady) => false,
                Err(e) => return Err(e),
            };

            if timeoutted {
                for h in hs {
                    self.connected.retain(|v| v.id != *h);
                }
            }
        }
        if timeoutted {
            self.timeout = None;
        }

        let (message, remote_id, hash) = match try_ready!(self.stream.poll()) {
            Some(Some(val)) => val,
            Some(None) =>
                if self.incoming.len() > 0 {
                    return Ok(Async::Ready(Some(self.incoming.pop().unwrap())));
                } else {
                    return Ok(Async::NotReady);
                },
            None => return Ok(Async::Ready(None)),
        };

        match message.typ {
            0x01 /* ping */ => {
                let incoming_message: PingMessage = match UntrustedRlp::new(&message.data).as_val() {
                    Ok(val) => val,
                    Err(_) =>
                        if self.incoming.len() > 0 {
                            return Ok(Async::Ready(Some(self.incoming.pop().unwrap())));
                        } else {
                            return Ok(Async::NotReady);
                        },
                };
                let typ = 0x02u8;
                let echo = hash;
                let to = incoming_message.from;
                let outgoing_message = PongMessage {
                    echo, to, timestamp: time::now_utc().to_timespec().sec as u64,
                };
                let data = rlp::encode(&outgoing_message).to_vec();

                // If sending pong is not available, drop.
                self.stream.start_send(DPTCodecMessage {
                    typ, data, addr: message.addr
                });<commit_msg>dpt: make no reachable endpoint as error<commit_after>//! Ethereum DPT protocol implementation

extern crate etcommon_bigint as bigint;
extern crate etcommon_crypto as hash;
extern crate etcommon_rlp as rlp;
extern crate sha3;
extern crate secp256k1;
#[macro_use]
extern crate futures;
extern crate tokio_io;
extern crate tokio_core;
extern crate time;

mod proto;
mod message;
mod util;

use message::*;
use proto::{DPTCodec, DPTCodecMessage};
use futures::future;
use futures::{Poll, Async, StartSend, AsyncSink, Future, Stream, Sink};
use tokio_io::{AsyncRead, AsyncWrite};
use tokio_io::codec::{Framed, Encoder, Decoder};
use tokio_core::reactor::{Timeout, Handle};
use tokio_core::net::{UdpSocket, UdpFramed};
use std::net::{IpAddr, SocketAddr, Ipv4Addr, Ipv6Addr};
use std::io;
use bigint::{H256, H512};
use rlp::UntrustedRlp;
use hash::SECP256K1;
use secp256k1::key::{PublicKey, SecretKey};
use util::pk2id;

fn retain_mut<T, F>(vec: &mut Vec<T>, mut f: F)
    where F: FnMut(&mut T) -> bool
{
    let len = vec.len();
    let mut del = 0;
    {
        let v = &mut **vec;

        for i in 0..len {
            if !f(&mut v[i]) {
                del += 1;
            } else if del > 0 {
                v.swap(i - del, i);
            }
        }
    }
    if del > 0 {
        vec.truncate(len - del);
    }
}

/// DPT message for requesting new peers or ping with timeout
pub enum DPTMessage {
    RequestNewPeer,
    Ping(Timeout),
}

/// DPT stream for sending DPT messages or receiving new peers
pub struct DPTStream {
    stream: UdpFramed<DPTCodec>,
    id: H512,
    connected: Vec<DPTNode>,
    timeout: Option<(Timeout, Vec<H512>)>,
    incoming: Vec<DPTNode>,
    address: IpAddr,
    udp_port: u16,
    tcp_port: u16,
}

#[derive(Debug, Clone)]
/// DPT node used by a DPT stream
pub struct DPTNode {
    pub address: IpAddr,
    pub tcp_port: u16,
    pub udp_port: u16,
    pub id: H512,
}

impl DPTNode {
    /// The TCP socket address of this node
    pub fn tcp_addr(&self) -> SocketAddr {
        SocketAddr::new(self.address, self.tcp_port)
    }

    /// The UDP socket address of this node
    pub fn udp_addr(&self) -> SocketAddr {
        SocketAddr::new(self.address, self.udp_port)
    }
}

impl DPTStream {
    /// Create a new DPT stream
    pub fn new(addr: &SocketAddr, handle: &Handle,
               secret_key: SecretKey,
               bootstrap_nodes: Vec<DPTNode>,
               tcp_port: u16) -> Result<Self, io::Error> {
        let id = pk2id(&match PublicKey::from_secret_key(&SECP256K1, &secret_key) {
            Ok(val) => val,
            Err(_) => return Err(io::Error::new(io::ErrorKind::Other, "converting pub key failed")),
        });
        Ok(Self {
            stream: UdpSocket::bind(addr, handle)?.framed(DPTCodec::new(secret_key)),
            id, connected: bootstrap_nodes, incoming: Vec::new(),
            timeout: None,
            address: addr.ip(), udp_port: addr.port(), tcp_port
        })
    }
}

impl Stream for DPTStream {
    type Item = DPTNode;
    type Error = io::Error;

    fn poll(&mut self) -> Poll<Option<Self::Item>, Self::Error> {
        let mut timeoutted = false;
        if self.timeout.is_some() {
            let &mut (ref mut timeout, ref hs) = self.timeout.as_mut().unwrap();
            timeoutted = match timeout.poll() {
                Ok(Async::Ready(())) => true,
                Ok(Async::NotReady) => false,
                Err(e) => return Err(e),
            };

            if timeoutted {
                for h in hs {
                    self.connected.retain(|v| v.id != *h);
                }
            }
        }
        if timeoutted {
            self.timeout = None;
        }

        let (message, remote_id, hash) = match try!(self.stream.poll()) {
            Async::Ready(Some(Some(val))) => val,
            Async::Ready(Some(None)) | Async::NotReady => {
                if self.incoming.len() > 0 {
                    return Ok(Async::Ready(Some(self.incoming.pop().unwrap())));
                } else if self.connected.len() == 0 {
                    return Err(io::Error::new(io::ErrorKind::Other, "no reachable DPT endpoint"));
                } else {
                    return Ok(Async::NotReady);
                }
            },
            Async::Ready(None) => return Ok(Async::Ready(None)),
        };

        match message.typ {
            0x01 /* ping */ => {
                let incoming_message: PingMessage = match UntrustedRlp::new(&message.data).as_val() {
                    Ok(val) => val,
                    Err(_) =>
                        if self.incoming.len() > 0 {
                            return Ok(Async::Ready(Some(self.incoming.pop().unwrap())));
                        } else {
                            return Ok(Async::NotReady);
                        },
                };
                let typ = 0x02u8;
                let echo = hash;
                let to = incoming_message.from;
                let outgoing_message = PongMessage {
                    echo, to, timestamp: time::now_utc().to_timespec().sec as u64,
                };
                let data = rlp::encode(&outgoing_message).to_vec();

                // If sending pong is not available, drop.
                self.stream.start_send(DPTCodecMessage {
                    typ, data, addr: message.addr
                });
{'fraction_non_alphanumeric': 0.12876190476190477, 'fraction_numerical': 0.011904761904761904, 'mean_word_length': 1.9104767184035476, 'pattern_counts': {'":': 0, '<': 25, '<?xml version=': 0, '>': 65, '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': '2679166', 'n_tokens_mistral': 3385, 'n_tokens_neox': 3223, 'n_words': 837}