<commit_before>extern crate erl_dist;
extern crate fibers;
extern crate futures;
extern crate handy_async;
extern crate clap;
extern crate eetf;

use std::io::{Error, ErrorKind};
use std::net::SocketAddr;
use clap::{App, Arg};
use fibers::Executor;
use fibers::net::TcpStream;
use futures::Future;

fn main() {
    let matches = App::new("handshake")
        .arg(Arg::with_name("EPMD_HOST").short("h").takes_value(true).default_value("127.0.0.1"))
        .arg(Arg::with_name("EPMD_PORT").short("p").takes_value(true).default_value("4369"))
        .arg(Arg::with_name("NODE_NAME").short("n").takes_value(true).default_value("foo"))
        .arg(Arg::with_name("COOKIE")
            .short("c")
            .takes_value(true)
            .default_value("WPKYDIOSJIMJUURLRUHV"))
        .arg(Arg::with_name("LOCAL_NODE")
            .short("l")
            .takes_value(true)
            .default_value("bar@127.0.0.1"))
        .arg(Arg::with_name("DESTINATION")
            .short("d")
            .takes_value(true)
            .default_value("foo"))
        .get_matches();
    let node_name = matches.value_of("NODE_NAME").unwrap().to_string();
    let local_node = matches.value_of("LOCAL_NODE").unwrap().to_string();
    let cookie = matches.value_of("COOKIE").unwrap().to_string();
    let epmd_host = matches.value_of("EPMD_HOST").unwrap();
    let epmd_port = matches.value_of("EPMD_PORT").unwrap();
    let epmd_addr: SocketAddr =
        format!("{}:{}", epmd_host, epmd_port).parse().expect("Invalid epmd address");
    let dest_proc = matches.value_of("DESTINATION").unwrap().to_string();

    let client = erl_dist::epmd::EpmdClient::new();
    let mut executor = fibers::InPlaceExecutor::new().unwrap();

    let epmd_ip = epmd_addr.ip();
    let local_node0 = local_node.to_string();
    let monitor = executor.spawn_monitor(TcpStream::connect(epmd_addr)
        .and_then(move |socket| client.get_node_info(socket, &node_name))
        .and_then(move |info| {
            if let Some(addr) = info.map(|i| SocketAddr::new(epmd_ip, i.port)) {
                fibers::net::TcpStream::connect(addr)
                    .and_then(move |socket| {
                        let local_node = erl_dist::epmd::NodeInfo::new(&local_node, 3333);
                        let handshake = erl_dist::handshake::Handshake::new(local_node, &cookie);
                        handshake.connect(socket)
                    })
                    .boxed()
            } else {
                futures::failed(Error::new(ErrorKind::NotFound, "target node is not found")).boxed()
            }
        })
        .and_then(move |(socket, flags)| {
            use futures::Sink;
            println!("# Distribution Flags: {:?}", flags);
            let tx = erl_dist::channel::sender(socket);
            let from_pid = eetf::Pid::new(local_node0, 0, 0, 0);
            let message =
                erl_dist::message::RegSend::new(from_pid, dest_proc, eetf::Atom::from("send_test"));
            tx.send(From::from(message))
        }));
    let r = executor.run_fiber(monitor).unwrap().expect("Handshake failed");
    println!("# RESULT: {:?}", r);
}<commit_msg>Refactor send_msg example<commit_after>//! Client Node Example.
//!
//! The node sends a message (atom) to the specified erlang node.
//!
//! # Usage Examples
//!
//! ```bash
//! $ cargo run --example send_msg -- --help
//! $ cargo run --example send_msg -- --peer foo --destination foo --cookie erlang_cookie -m hello
//! ```
extern crate erl_dist;
extern crate fibers;
extern crate futures;
extern crate clap;
extern crate eetf;

use std::io::{Error, ErrorKind};
use std::net::SocketAddr;
use clap::{App, Arg};
use fibers::{Executor, InPlaceExecutor};
use fibers::net::TcpStream;
use futures::{Sink, Future};
use futures::future::Either;
use erl_dist::{EpmdClient, Message, Handshake};
use erl_dist::channel;

fn main() {
    let matches = App::new("send_msg")
        .arg(Arg::with_name("EPMD_HOST").short("h").takes_value(true).default_value("127.0.0.1"))
        .arg(Arg::with_name("EPMD_PORT").short("p").takes_value(true).default_value("4369"))
        .arg(Arg::with_name("PEER_NAME").long("peer").takes_value(true).default_value("foo"))
        .arg(Arg::with_name("COOKIE")
            .long("cookie")
            .takes_value(true)
            .default_value("WPKYDIOSJIMJUURLRUHV"))
        .arg(Arg::with_name("SELF_NODE")
            .long("self")
            .takes_value(true)
            .default_value("bar@localhost"))
        .arg(Arg::with_name("DESTINATION")
            .short("d")
            .long("destination")
            .takes_value(true)
            .default_value("foo"))
        .arg(Arg::with_name("MESSAGE")
            .short("m")
            .long("message")
            .takes_value(true)
            .default_value("hello_world"))
        .get_matches();
    let peer_name = matches.value_of("PEER_NAME").unwrap().to_string();
    let self_node = matches.value_of("SELF_NODE").unwrap().to_string();
    let cookie = matches.value_of("COOKIE").unwrap().to_string();
    let epmd_host = matches.value_of("EPMD_HOST").unwrap();
    let epmd_port = matches.value_of("EPMD_PORT").unwrap();
    let epmd_addr: SocketAddr =
        format!("{}:{}", epmd_host, epmd_port).parse().expect("Invalid epmd address");
    let dest_proc = matches.value_of("DESTINATION").unwrap().to_string();
    let message = matches.value_of("MESSAGE").unwrap().to_string();

    let self_node0 = self_node.to_string();
    let mut executor = InPlaceExecutor::new().unwrap();
    let monitor = executor.spawn_monitor(TcpStream::connect(epmd_addr.clone())
        .and_then(move |epmd_socket| {
            // Gets peer node information from the EPMD
            EpmdClient::new().get_node_info(epmd_socket, &peer_name)
        })
        .and_then(move |info| {
            if let Some(addr) = info.map(|i| SocketAddr::new(epmd_addr.ip(), i.port)) {
                // Executes the client side handshake
                Either::A(TcpStream::connect(addr).and_then(move |socket| {
                    let handshake = Handshake::new(&self_node, &cookie);
                    handshake.connect(socket)
                }))
            } else {
                Either::B(futures::failed(Error::new(ErrorKind::NotFound,
                                                     "target node is not found")))
            }
        })
        .and_then(move |peer| {
            // Sends a message to the peer node
            println!("# Connected: {}", peer.name);
            println!("# Distribution Flags: {:?}", peer.flags);
            let tx = channel::sender(peer.stream);
            let from_pid = eetf::Pid::new(self_node0, 0, 0, 0);
            let atom = eetf::Atom::from(message);
            let message = Message::reg_send(from_pid, dest_proc, atom);
            println!("# Send: {:?}", message);
            tx.send(message)
        }));
    let _ = executor.run_fiber(monitor).unwrap().expect("Failed");
    println!("# DONE");
}
{'fraction_non_alphanumeric': 0.16488222698072805, 'fraction_numerical': 0.005710206995003569, 'mean_word_length': 2.706878306878307, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 3, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 2, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '768441', 'n_tokens_mistral': 2277, 'n_tokens_neox': 2221, 'n_words': 379}