<commit_before>// Copyright 2018 Grove Enterprises LLC
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

use std::fs::File;
use std::io::prelude::*;
use std::io::{Error, ErrorKind};
use std::time::Duration;
use std::thread;

extern crate clap;
extern crate datafusion;
extern crate etcd;
extern crate futures;
extern crate futures_timer;
extern crate hyper;
extern crate serde;
extern crate serde_json;
extern crate tokio_core;
extern crate uuid;

use clap::{Arg, App};
use etcd::Client;
use etcd::kv;
use datafusion::exec::*;
use futures::future::{ok, loop_fn, Future, Loop};
use futures::Stream;
use hyper::{Method, StatusCode, Chunk};
use hyper::client::HttpConnector;
use hyper::header::{ContentLength};
use hyper::server::{Http, Request, Response, Service};
use tokio_core::reactor::Core;
use uuid::Uuid;

fn main() {

    let matches = App::new("DataFusion Worker Node")
        .version("0.1.4") //TODO get dynamically based on crate version
        .arg(Arg::with_name("ETCD")
            .help("etcd endpoints")
            .short("e")
            .long("etcd")
            .value_name("URL")

            .takes_value(true))
        .arg(Arg::with_name("BIND")
            .short("b")
            .long("bind")
            .help("IP address and port to bind to")
            .default_value("0.0.0.0:8080")
            .takes_value(true))
        .arg(Arg::with_name("WEBROOT")
            .short("w")
            .long("webroot")
            .help("Location of HTML files")
            .default_value("./src/bin/worker/")
            .takes_value(true))
        .get_matches();

    let uuid = Uuid::new_v5(&uuid::NAMESPACE_DNS, "datafusion");
    let bind_addr_str = matches.value_of("BIND").unwrap().to_string();

    let bind_addr = bind_addr_str.parse().unwrap();
    let www_root = matches.value_of("WEBROOT").unwrap().to_string();

    let etcd_endpoints = matches.value_of("ETCD").unwrap();

    // create futures event loop
    let mut core = Core::new().unwrap();
    let handle = core.handle();

    println!("Worker {} listening on {} and serving content from {}", uuid, bind_addr, www_root);

    let _ = Http::new()
        .bind(&bind_addr, move|| Ok(Worker { www_root: www_root.clone() })).unwrap();

    // start a loop to register with etcd every 5 seconds with a ttl of 10 seconds
    let etcd = Client::new(&handle, &[etcd_endpoints], None).unwrap();
    let heartbeat_loop = loop_fn(Membership::new(etcd, uuid, bind_addr_str), |client| {
        client.register()
            .and_then(|(client, done)| {
                if done {
                    Ok(Loop::Break(client))
                } else {
                    Ok(Loop::Continue(client))
                }
            })
    });
    core.run(heartbeat_loop).unwrap();

}

/// Worker struct to store state
struct Worker {<commit_msg>clean up error handling in worker<commit_after>// Copyright 2018 Grove Enterprises LLC
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

use std::fs::File;
use std::io::prelude::*;
use std::io::{Error, ErrorKind};
use std::time::Duration;
use std::thread;

extern crate clap;
extern crate datafusion;
extern crate etcd;
extern crate futures;
extern crate futures_timer;
extern crate hyper;
extern crate serde;
extern crate serde_json;
extern crate tokio_core;
extern crate uuid;

use clap::{Arg, App};
use etcd::Client;
use etcd::kv;
use datafusion::exec::*;
use futures::future::{ok, loop_fn, Future, Loop};
use futures::Stream;
use hyper::{Method, StatusCode, Chunk};
use hyper::client::HttpConnector;
use hyper::header::{ContentLength};
use hyper::server::{Http, Request, Response, Service};
use tokio_core::reactor::Core;
use uuid::Uuid;

fn main() {

    let matches = App::new("DataFusion Worker Node")
        .version("0.1.4") //TODO get dynamically based on crate version
        .arg(Arg::with_name("ETCD")
            .help("etcd endpoints")
            .short("e")
            .long("etcd")
            .value_name("URL")
            .required(true)
            .takes_value(true))
        .arg(Arg::with_name("BIND")
            .short("b")
            .long("bind")
            .help("IP address and port to bind to")
            .default_value("0.0.0.0:8080")
            .takes_value(true))
        .arg(Arg::with_name("WEBROOT")
            .short("w")
            .long("webroot")
            .help("Location of HTML files")
            .default_value("./src/bin/worker/")
            .takes_value(true))
        .get_matches();

    let uuid = Uuid::new_v5(&uuid::NAMESPACE_DNS, "datafusion");
    let bind_addr_str = matches.value_of("BIND").unwrap().to_string();

    let bind_addr = bind_addr_str.parse().unwrap();
    let www_root = matches.value_of("WEBROOT").unwrap().to_string();

    let etcd_endpoints = matches.value_of("ETCD").unwrap();

    // create futures event loop
    let mut core = Core::new().unwrap();
    let handle = core.handle();

    println!("Worker {} listening on {} and serving content from {}", uuid, bind_addr, www_root);

    let _ = Http::new()
        .bind(&bind_addr, move|| Ok(Worker { www_root: www_root.clone() })).unwrap();

    // start a loop to register with etcd every 5 seconds with a ttl of 10 seconds
    match Client::new(&handle, &[etcd_endpoints], None) {
        Ok(etcd) => {
            let heartbeat_loop = loop_fn(Membership::new(etcd, uuid, bind_addr_str), |client| {
                client.register()
                    .and_then(|(client, done)| {
                        if done {
                            Ok(Loop::Break(client))
                        } else {
                            Ok(Loop::Continue(client))
                        }
                    })
            });
            match core.run(heartbeat_loop) {
                Ok(_) => println!("Heartbeat loop finished"),
                Err(e) => println!("Heartbeat loop failed: {:?}", e),
            }

        }
        Err(e) => println!("Failed to connect to etcd: {:?}", e)
    }

}

/// Worker struct to store state
struct Worker {
{'fraction_non_alphanumeric': 0.14084306465394475, 'fraction_numerical': 0.006550840216462546, 'mean_word_length': 2.66544885177453, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 7, 'https://': 0, 'lorem ipsum': 0, 'www.': 10, 'xml': 0}, 'pii_count': 2, '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': '884222', 'n_tokens_mistral': 2136, 'n_tokens_neox': 2077, 'n_words': 603}