<commit_before>#![feature(plugin)]
#![feature(scoped)]
#![plugin(quickcheck_macros)]
#![feature(convert)]
#[macro_use] extern crate log;
extern crate env_logger;
extern crate yak_client;
extern crate capnp;
extern crate log4rs;
#[cfg(test)]
extern crate quickcheck;

use std::default::Default;
use std::net::{TcpListener, TcpStream};
use std::thread;
use std::io::{self,Read,Write};
use std::fmt;
use std::sync::{Arc,Mutex};
use std::collections::HashMap;
use std::clone::Clone;
use std::error::Error;

use yak_client::{WireProtocol,Request,Response,Operation,Datum,YakError};

#[macro_use] mod store;
mod mem_store;

#[derive(Debug)]
enum ServerError {
  CapnpError(capnp::Error),
  CapnpNotInSchema(capnp::NotInSchema),
  IoError(std::io::Error),
  DownstreamError(YakError)
}

impl fmt::Display for ServerError {
  fn fmt(&self, f: &mut fmt::Formatter) -> Result<(), fmt::Error> {
    match self {
      &ServerError::CapnpError(ref e) => e.fmt(f),
      &ServerError::CapnpNotInSchema(ref e) => e.fmt(f),
      &ServerError::IoError(ref e) => e.fmt(f),
      &ServerError::DownstreamError(ref e) => e.fmt(f)
    }
  }
}

impl Error for ServerError {
  fn description(&self) -> &str {
    match self {
      &ServerError::CapnpError(ref e) => e.description(),
      &ServerError::CapnpNotInSchema(ref e) => e.description(),
      &ServerError::IoError(ref e) => e.description(),
      &ServerError::DownstreamError(ref e) => e.description()
    }
  }
}

struct DownStream<S: Read+Write> {
  protocol: Arc<Mutex<WireProtocol<S>>>,
}

impl <S: ::std::fmt::Debug + Read + Write> ::std::fmt::Debug for DownStream<S>
 where S: ::std::fmt::Debug + 'static {
    fn fmt(&self, fmt: &mut ::std::fmt::Formatter) -> ::std::fmt::Result {
        match self {
            &DownStream { protocol: ref proto } =>
            fmt.debug_struct("DownStream").field("protocol",
                                                     (proto)).finish(),
        }
    }
}

impl<S> Clone for DownStream<S> where S: Read+Write {
  fn clone(&self) -> DownStream<S> {
    DownStream { protocol: self.protocol.clone() }
  }
}

static LOG_FILE: &'static str = "log.toml";

pub fn main() {
  if let Err(e) = log4rs::init_file(LOG_FILE, Default::default()) {
    panic!("Could not init logger from file {}: {}", LOG_FILE, e);
  }

  match do_run() {
    Ok(()) => info!("Terminated normally"),
    Err(e) => panic!("Failed: {}", e),
  }
}

fn do_run() -> Result<(), ServerError> {
  let mut a = std::env::args().skip(1);
  let local : String = a.next().unwrap();
  let next = match a.next() {
      Some(ref addr) => Some(try!(DownStream::new(addr))),
      None => None
  };

  let listener = TcpListener::bind(local.as_str()).unwrap();
  info!("listening started on {}, ready to accept", local);
  let store = mem_store::MemStore::new();
  for stream in listener.incoming() {
    let next = next.clone();
    let store = store.clone();
    let sock = stream.unwrap();
    let peer = sock.peer_addr().unwrap();
    let _ = try!(thread::Builder::new().name(format!("client-{}", peer)).spawn(move || {
	debug!("Accept stream from {:?}", peer);
        match Session::new(peer, sock, store, next).process_requests() {
          Err(e) => report_errors(&e),
          _ => ()
        }
      }));
  }
  Ok(())
}<commit_msg>Change thread name so that client ip fits in ps output.<commit_after>#![feature(plugin)]
#![feature(scoped)]
#![plugin(quickcheck_macros)]
#![feature(convert)]
#[macro_use] extern crate log;
extern crate env_logger;
extern crate yak_client;
extern crate capnp;
extern crate log4rs;
#[cfg(test)]
extern crate quickcheck;

use std::default::Default;
use std::net::{TcpListener, TcpStream};
use std::thread;
use std::io::{self,Read,Write};
use std::fmt;
use std::sync::{Arc,Mutex};
use std::collections::HashMap;
use std::clone::Clone;
use std::error::Error;

use yak_client::{WireProtocol,Request,Response,Operation,Datum,YakError};

#[macro_use] mod store;
mod mem_store;

#[derive(Debug)]
enum ServerError {
  CapnpError(capnp::Error),
  CapnpNotInSchema(capnp::NotInSchema),
  IoError(std::io::Error),
  DownstreamError(YakError)
}

impl fmt::Display for ServerError {
  fn fmt(&self, f: &mut fmt::Formatter) -> Result<(), fmt::Error> {
    match self {
      &ServerError::CapnpError(ref e) => e.fmt(f),
      &ServerError::CapnpNotInSchema(ref e) => e.fmt(f),
      &ServerError::IoError(ref e) => e.fmt(f),
      &ServerError::DownstreamError(ref e) => e.fmt(f)
    }
  }
}

impl Error for ServerError {
  fn description(&self) -> &str {
    match self {
      &ServerError::CapnpError(ref e) => e.description(),
      &ServerError::CapnpNotInSchema(ref e) => e.description(),
      &ServerError::IoError(ref e) => e.description(),
      &ServerError::DownstreamError(ref e) => e.description()
    }
  }
}

struct DownStream<S: Read+Write> {
  protocol: Arc<Mutex<WireProtocol<S>>>,
}

impl <S: ::std::fmt::Debug + Read + Write> ::std::fmt::Debug for DownStream<S>
 where S: ::std::fmt::Debug + 'static {
    fn fmt(&self, fmt: &mut ::std::fmt::Formatter) -> ::std::fmt::Result {
        match self {
            &DownStream { protocol: ref proto } =>
            fmt.debug_struct("DownStream").field("protocol",
                                                     (proto)).finish(),
        }
    }
}

impl<S> Clone for DownStream<S> where S: Read+Write {
  fn clone(&self) -> DownStream<S> {
    DownStream { protocol: self.protocol.clone() }
  }
}

static LOG_FILE: &'static str = "log.toml";

pub fn main() {
  if let Err(e) = log4rs::init_file(LOG_FILE, Default::default()) {
    panic!("Could not init logger from file {}: {}", LOG_FILE, e);
  }

  match do_run() {
    Ok(()) => info!("Terminated normally"),
    Err(e) => panic!("Failed: {}", e),
  }
}

fn do_run() -> Result<(), ServerError> {
  let mut a = std::env::args().skip(1);
  let local : String = a.next().unwrap();
  let next = match a.next() {
      Some(ref addr) => Some(try!(DownStream::new(addr))),
      None => None
  };

  let listener = TcpListener::bind(local.as_str()).unwrap();
  info!("listening started on {}, ready to accept", local);
  let store = mem_store::MemStore::new();
  for stream in listener.incoming() {
    let next = next.clone();
    let store = store.clone();
    let sock = stream.unwrap();
    let peer = sock.peer_addr().unwrap();
    let _ = try!(thread::Builder::new().name(format!("C{}", peer)).spawn(move || {
	debug!("Accept stream from {:?}", peer);
        match Session::new(peer, sock, store, next).process_requests() {
          Err(e) => report_errors(&e),
          _ => ()
        }
      }));
  }
  Ok(())
}
{'fraction_non_alphanumeric': 0.18755662941709453, 'fraction_numerical': 0.000906070673512534, 'mean_word_length': 3.7819494584837545, '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': 4, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2698186', 'n_tokens_mistral': 2220, 'n_tokens_neox': 2246, 'n_words': 492}