<commit_before>// Copyright 2015 Copyright (c) 2015 <NAME> (<EMAIL>)
//
// Licensed under the MIT license LICENSE or <http://opensource.org/licenses/MIT>
// This file may not be copied, modified, or distributed except according to those terms.

use std::rc::Rc;
use std::io;

use byteorder::{ BigEndian, WriteBytesExt, ReadBytesExt };

use mio;

use EventLoop;
use Message;
use transport::Connection;
use global;

// An endpoint is responsible for handshaking with its peer and transfering raw messages over a connection.
// That means send/receive size prefix and then message payload
// according to the connection readiness and the requested operation progress
pub struct Endpoint {
    token: mio::Token, 
    addr: Option<String>,
    handshake_state: Option<HandshakeEndpointState>,
    connected_state: Option<ConnectedEndpointState>
}

pub enum SendStatus {
    Postponed(Rc<Message>), // Message can't be sent at the moment : Handshake in progress or would block
    Completed,              // Message has been successfully sent
    InProgress              // Message has been partially sent, will finish later
}

pub enum RecvStatus {
    Postponed,              // Message can't be read at the moment : Handshake in progress or would block
    Completed(Message),     // Message has been successfully read
    InProgress              // Message has been partially read, will finish later
}

impl Endpoint {

    pub fn new(
        token: mio::Token, 
        addr: Option<String>, 
        proto_ids: (u16, u16), 
        connection: Box<Connection>) -> Endpoint {

        let handshake = HandshakeEndpointState::new(token, proto_ids, connection);

        Endpoint {
            token: token,
            addr: addr,
            handshake_state: Some(handshake),
            connected_state: None
        }
    }

    pub fn open(&self, event_loop: &mut EventLoop) -> io::Result<()> {
        self.handshake_state.as_ref().map_or_else(
            || Err(global::other_io_error("cannot open pipe after handshake step !")),
            |hs| hs.open(event_loop))
    }

    pub fn close(&self, event_loop: &mut EventLoop) -> io::Result<()> {
        if self.handshake_state.is_some() {
            self.handshake_state.as_ref().unwrap().close(event_loop)
        } else if self.connected_state.is_some() {
            self.connected_state.as_ref().unwrap().close(event_loop)
        } else {
            Ok(())
        }
    }

    pub fn addr(self) -> Option<String> {
        self.addr
    }

    pub fn ready(&mut self, event_loop: &mut EventLoop, events: mio::EventSet) -> io::Result<(bool, Option<Message>)> {

        if events.is_hup() {
            return Err(io::Error::new(io::ErrorKind::ConnectionReset, "event: hup"));
        }

        if events.is_error() {
            return Err(io::Error::new(io::ErrorKind::ConnectionAborted, "event: error"));
        }

        if self.handshake_state.is_some() {
            let mut handshake = self.handshake_state.take().unwrap();
            if try!(handshake.ready(event_loop, events)) {
                let connection = handshake.connection();
                let connected = ConnectedEndpointState::new(self.token, connection);

                try!(connected.open(event_loop));
                self.connected_state = Some(connected)
            } else {
                try!(handshake.reopen(event_loop));
                self.handshake_state = Some(handshake);
            }

            return Ok((false, None));
        } 

        self.connected_state.as_mut().map_or_else(
            || Err(global::other_io_error("ready notification while pipe is dead")),
            |mut connected| connected.ready(event_loop, events))
    }

    pub fn send(&mut self, msg: Rc<Message>) -> io::Result<SendStatus> {
        if self.handshake_state.is_some() {
            return Ok(SendStatus::Postponed(msg));
        }

        self.connected_state.as_mut().map_or_else(
            || Err(global::other_io_error("cannot send when pipe is dead")),
            |mut connected| connected.send(msg))        
    }

    pub fn cancel_sending(&mut self) {
        self.connected_state.as_mut().map(|mut connected| connected.cancel_sending());
    }

    pub fn recv(&mut self) -> io::Result<RecvStatus> {
        if self.handshake_state.is_some() {
            return Ok(RecvStatus::Postponed);
        }

        self.connected_state.as_mut().map_or_else(
            || Err(global::other_io_error("cannot receive when pipe is dead")),
            |mut connected| connected.recv())       
    }

    pub fn cancel_receiving(&mut self) {
        self.connected_state.as_mut().map(|mut connected| connected.cancel_receiving());
    }
}


struct HandshakeEndpointState {
    token: mio::Token,
    protocol_id: u16,
    protocol_peer_id: u16,
    connection: Box<Connection>,
    sent: bool,
    received: bool
}

impl HandshakeEndpointState {

    fn new(token: mio::Token, proto_ids: (u16, u16), connection: Box<Connection>) -> HandshakeEndpointState {
        let (protocol_id, protocol_peer_id) = proto_ids;

        HandshakeEndpointState { 
            token: token,
            protocol_id: protocol_id,
            protocol_peer_id: protocol_peer_id,
            connection: connection,
            sent: false, 
            received: false }
    }

    fn open(&self, event_loop: &mut EventLoop) -> io::Result<()> {
        event_loop.register(
            self.connection.as_evented(), 
            self.token, 
            mio::EventSet::all(), 
            mio::PollOpt::oneshot())
    }

    fn reopen(&self, event_loop: &mut EventLoop) -> io::Result<()> {
        event_loop.reregister(
            self.connection.as_evented(), 
            self.token, 
            mio::EventSet::all(), 
            mio::PollOpt::oneshot())
    }
<commit_msg>Uses edge opt instead of onehsot during handshake to see if appveyor build likes it better<commit_after>// Copyright 2015 Copyright (c) 2015 <NAME> (<EMAIL>)
//
// Licensed under the MIT license LICENSE or <http://opensource.org/licenses/MIT>
// This file may not be copied, modified, or distributed except according to those terms.

use std::rc::Rc;
use std::io;

use byteorder::{ BigEndian, WriteBytesExt, ReadBytesExt };

use mio;

use EventLoop;
use Message;
use transport::Connection;
use global;

// An endpoint is responsible for handshaking with its peer and transfering raw messages over a connection.
// That means send/receive size prefix and then message payload
// according to the connection readiness and the requested operation progress
pub struct Endpoint {
    token: mio::Token, 
    addr: Option<String>,
    handshake_state: Option<HandshakeEndpointState>,
    connected_state: Option<ConnectedEndpointState>
}

pub enum SendStatus {
    Postponed(Rc<Message>), // Message can't be sent at the moment : Handshake in progress or would block
    Completed,              // Message has been successfully sent
    InProgress              // Message has been partially sent, will finish later
}

pub enum RecvStatus {
    Postponed,              // Message can't be read at the moment : Handshake in progress or would block
    Completed(Message),     // Message has been successfully read
    InProgress              // Message has been partially read, will finish later
}

impl Endpoint {

    pub fn new(
        token: mio::Token, 
        addr: Option<String>, 
        proto_ids: (u16, u16), 
        connection: Box<Connection>) -> Endpoint {

        let handshake = HandshakeEndpointState::new(token, proto_ids, connection);

        Endpoint {
            token: token,
            addr: addr,
            handshake_state: Some(handshake),
            connected_state: None
        }
    }

    pub fn open(&self, event_loop: &mut EventLoop) -> io::Result<()> {
        self.handshake_state.as_ref().map_or_else(
            || Err(global::other_io_error("cannot open pipe after handshake step !")),
            |hs| hs.open(event_loop))
    }

    pub fn close(&self, event_loop: &mut EventLoop) -> io::Result<()> {
        if self.handshake_state.is_some() {
            self.handshake_state.as_ref().unwrap().close(event_loop)
        } else if self.connected_state.is_some() {
            self.connected_state.as_ref().unwrap().close(event_loop)
        } else {
            Ok(())
        }
    }

    pub fn addr(self) -> Option<String> {
        self.addr
    }

    pub fn ready(&mut self, event_loop: &mut EventLoop, events: mio::EventSet) -> io::Result<(bool, Option<Message>)> {

        if events.is_hup() {
            return Err(io::Error::new(io::ErrorKind::ConnectionReset, "event: hup"));
        }

        if events.is_error() {
            return Err(io::Error::new(io::ErrorKind::ConnectionAborted, "event: error"));
        }

        if self.handshake_state.is_some() {
            let mut handshake = self.handshake_state.take().unwrap();
            if try!(handshake.ready(event_loop, events)) {
                let connection = handshake.connection();
                let connected = ConnectedEndpointState::new(self.token, connection);

                try!(connected.open(event_loop));
                self.connected_state = Some(connected)
            } else {
                try!(handshake.reopen(event_loop));
                self.handshake_state = Some(handshake);
            }

            return Ok((false, None));
        } 

        self.connected_state.as_mut().map_or_else(
            || Err(global::other_io_error("ready notification while pipe is dead")),
            |mut connected| connected.ready(event_loop, events))
    }

    pub fn send(&mut self, msg: Rc<Message>) -> io::Result<SendStatus> {
        if self.handshake_state.is_some() {
            return Ok(SendStatus::Postponed(msg));
        }

        self.connected_state.as_mut().map_or_else(
            || Err(global::other_io_error("cannot send when pipe is dead")),
            |mut connected| connected.send(msg))        
    }

    pub fn cancel_sending(&mut self) {
        self.connected_state.as_mut().map(|mut connected| connected.cancel_sending());
    }

    pub fn recv(&mut self) -> io::Result<RecvStatus> {
        if self.handshake_state.is_some() {
            return Ok(RecvStatus::Postponed);
        }

        self.connected_state.as_mut().map_or_else(
            || Err(global::other_io_error("cannot receive when pipe is dead")),
            |mut connected| connected.recv())       
    }

    pub fn cancel_receiving(&mut self) {
        self.connected_state.as_mut().map(|mut connected| connected.cancel_receiving());
    }
}


struct HandshakeEndpointState {
    token: mio::Token,
    protocol_id: u16,
    protocol_peer_id: u16,
    connection: Box<Connection>,
    sent: bool,
    received: bool
}

impl HandshakeEndpointState {

    fn new(token: mio::Token, proto_ids: (u16, u16), connection: Box<Connection>) -> HandshakeEndpointState {
        let (protocol_id, protocol_peer_id) = proto_ids;

        HandshakeEndpointState { 
            token: token,
            protocol_id: protocol_id,
            protocol_peer_id: protocol_peer_id,
            connection: connection,
            sent: false, 
            received: false }
    }

    fn open(&self, event_loop: &mut EventLoop) -> io::Result<()> {
        event_loop.register(
            self.connection.as_evented(), 
            self.token, 
            mio::EventSet::all(), 
            mio::PollOpt::edge())
    }

    fn reopen(&self, event_loop: &mut EventLoop) -> io::Result<()> {
        event_loop.reregister(
            self.connection.as_evented(), 
            self.token, 
            mio::EventSet::all(), 
            mio::PollOpt::edge())
    }

{'fraction_non_alphanumeric': 0.1157885748441636, 'fraction_numerical': 0.0034155921782939116, 'mean_word_length': 2.4447058823529413, 'pattern_counts': {'":': 0, '<': 45, '<?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': '5647976', 'n_tokens_mistral': 3438, 'n_tokens_neox': 3297, 'n_words': 821}