<commit_before>#![cfg_attr(feature="clippy", feature(plugin))]
#![cfg_attr(feature="clippy", plugin(clippy))]

#[macro_use]
extern crate mysql;
extern crate zmq;

mod transaction;
mod mapper;
#[macro_use]
mod macros;

use mapper::Mapper;
use std::sync::{Arc, Mutex, mpsc};
use std::thread;
use transaction::Transaction;

const WRITE_THREADS_COUNT: usize = 4;
const APPROVE_THREADS_COUNT: usize = 4;
const ZMQ_URI: &str = "tcp://172.16.17.32:5556";
const MYSQL_URI: &str = "mysql://root:password@127.0.0.1:\
                         3306/iota?prefer_socket=false";

fn main() {
  let pool = mysql::Pool::new(MYSQL_URI).unwrap();
  let ctx = zmq::Context::new();
  let socket = ctx.socket(zmq::SUB).unwrap();
  let (write_tx, write_rx) = mpsc::channel::<String>();
  let (approve_tx, approve_rx) = mpsc::channel::<Vec<u64>>();
  socket.connect(ZMQ_URI).unwrap();
  socket.set_subscribe(b"tx ").unwrap();
  spawn_write_workers(&pool, write_rx, &approve_tx);
  spawn_approve_workers(&pool, approve_rx);
  zmq_listen(&socket, &write_tx);
}

fn zmq_listen(socket: &zmq::Socket, tx: &mpsc::Sender<String>) {
  loop {
    match socket.recv_string(0) {
      Ok(Ok(string)) => tx.send(string).expect("Thread communication failure"),
      Ok(Err(err)) => eprintln!("Unexpected byte sequence: {:?}", err),
      Err(err) => eprintln!("{}", err),
    }
  }
}<commit_msg>Improved error messages<commit_after>#![cfg_attr(feature="clippy", feature(plugin))]
#![cfg_attr(feature="clippy", plugin(clippy))]

#[macro_use]
extern crate mysql;
extern crate zmq;

mod transaction;
mod mapper;
#[macro_use]
mod macros;

use mapper::Mapper;
use std::sync::{Arc, Mutex, mpsc};
use std::thread;
use transaction::Transaction;

const WRITE_THREADS_COUNT: usize = 4;
const APPROVE_THREADS_COUNT: usize = 4;
const ZMQ_URI: &str = "tcp://8172.16.17.32:5556";
const MYSQL_URI: &str = "mysql://root:password@127.0.0.1:\
                         3306/iota?prefer_socket=false";

fn main() {
  let pool = mysql::Pool::new(MYSQL_URI).expect("MySQL connect failure");
  let ctx = zmq::Context::new();
  let socket = ctx.socket(zmq::SUB).expect("ZMQ socket create failure");
  let (write_tx, write_rx) = mpsc::channel::<String>();
  let (approve_tx, approve_rx) = mpsc::channel::<Vec<u64>>();
  socket.connect(ZMQ_URI).expect("ZMQ socket connect failure");
  socket.set_subscribe(b"tx ").expect("ZMQ subscribe failure");
  spawn_write_workers(&pool, write_rx, &approve_tx);
  spawn_approve_workers(&pool, approve_rx);
  zmq_listen(&socket, &write_tx);
}

fn zmq_listen(socket: &zmq::Socket, tx: &mpsc::Sender<String>) {
  loop {
    match socket.recv_string(0) {
      Ok(Ok(string)) => tx.send(string).expect("Thread communication failure"),
      Ok(Err(err)) => eprintln!("Unexpected byte sequence: {:?}", err),
      Err(err) => eprintln!("{}", err),
    }
  }
}
{'fraction_non_alphanumeric': 0.18445229681978798, 'fraction_numerical': 0.020141342756183747, 'mean_word_length': 5.333333333333333, 'pattern_counts': {'":': 0, '<': 11, '<?xml version=': 0, '>': 17, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 4, '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': '1912246', 'n_tokens_mistral': 1072, 'n_tokens_neox': 1043, 'n_words': 200}