<commit_before>#include <vector>
#include <boost/lexical_cast.hpp>
#include <boost/bind.hpp>
#include <boost/foreach.hpp>
#include <boost/thread.hpp>
#include "zqrpc/ZSocket.hpp"


namespace zqrpc {
struct Worker {
	typedef std::vector<std::string> FramesT;
	zmq::context_t* context_;

	Worker(zmq::context_t* context) : context_(context) {}
	~Worker() {}
	void operator() () {
		zqrpc::ZSocket socket(context_,ZMQ_DEALER,boost::lexical_cast<std::string>(boost::this_thread::get_id()));
		socket.connect(ZQRPC_INPROC_WORKER);
		socket.SetLinger(0);
		typedef std::deque<std::string> FramesT;
		FramesT frames;
		while (1) {
			frames.clear();
			frames = socket.BlockingRecv<FramesT>();
			BOOST_FOREACH(std::string s, frames) { std::cerr << "## " << s << std::endl; }
			socket.Send<FramesT>(frames);
		}
	}
};
}


int main(int argc, char *argv[])
{
	google::InitGoogleLogging(argv[0]);
	zmq::context_t context;
	zqrpc::ZSocket rpc_frontend_(&context,ZMQ_ROUTER,"ROUTER");
	zqrpc::ZSocket rpc_backend_(&context,ZMQ_DEALER,"DEALER");
	boost::thread_group threads;
	rpc_frontend_.bind("tcp://127.0.0.1:9089");
	rpc_backend_.bind(ZQRPC_INPROC_WORKER);
	std::size_t noof_threads=2;
	for(std::size_t i = 0; i < noof_threads; ++i) {
		zqrpc::Worker w(&context);
		threads.create_thread<zqrpc::Worker>(w);
	}
	rpc_frontend_.ProxyTo( rpc_backend_);
	try {
	} catch (zqrpc::ZError& e) {
		DLOG(INFO) << "ZError THROWN" << std::endl;
	} catch (zqrpc::RetryException& e) {
		DLOG(INFO) << "Retry Exception THROWN" << std::endl;
	} catch (zmq::error_t& e) {
		DLOG(INFO) << "ZMQ ZERROR THROWN : " << e.what() << std::endl;
	}
	threads.join_all();<commit_msg>Wed Jul 23 15:15:56 IST 2014<commit_after>#include <vector>
#include <boost/lexical_cast.hpp>
#include <boost/bind.hpp>
#include <boost/foreach.hpp>
#include <boost/thread.hpp>
#include "zqrpc/ZSocket.hpp"


namespace zqrpc {
struct Worker {
	typedef std::vector<std::string> FramesT;
	zmq::context_t* context_;

	Worker(zmq::context_t* context) : context_(context) {}
	~Worker() {}
	void operator() () {
		zqrpc::ZSocket socket(context_,ZMQ_DEALER,boost::lexical_cast<std::string>(boost::this_thread::get_id()));
		socket.connect(ZQRPC_INPROC_WORKER);
		socket.SetLinger(0);
		typedef std::deque<std::string> FramesT;
		FramesT frames;
		while (1) {
			frames.clear();
			frames = socket.BlockingRecv<FramesT>();
			BOOST_FOREACH(std::string s, frames) { std::cerr << "## " << s.length() << " " << s << std::endl; }
			socket.Send<FramesT>(frames);
		}
	}
};
}


int main(int argc, char *argv[])
{
	google::InitGoogleLogging(argv[0]);
	zmq::context_t context;
	zqrpc::ZSocket rpc_frontend_(&context,ZMQ_ROUTER,"ROUTER");
	zqrpc::ZSocket rpc_backend_(&context,ZMQ_DEALER,"DEALER");
	boost::thread_group threads;
	rpc_frontend_.bind("tcp://127.0.0.1:9089");
	rpc_backend_.bind(ZQRPC_INPROC_WORKER);
	std::size_t noof_threads=2;
	for(std::size_t i = 0; i < noof_threads; ++i) {
		zqrpc::Worker w(&context);
		threads.create_thread<zqrpc::Worker>(w);
	}
	rpc_frontend_.ProxyTo( rpc_backend_);
	try {
	} catch (zqrpc::ZError& e) {
		DLOG(INFO) << "ZError THROWN" << std::endl;
	} catch (zqrpc::RetryException& e) {
		DLOG(INFO) << "Retry Exception THROWN" << std::endl;
	} catch (zmq::error_t& e) {
		DLOG(INFO) << "ZMQ ZERROR THROWN : " << e.what() << std::endl;
	}
	threads.join_all();
{'fraction_non_alphanumeric': 0.1864864864864865, 'fraction_numerical': 0.012612612612612612, 'mean_word_length': 5.939583333333333, 'pattern_counts': {'":': 0, '<': 71, '<?xml version=': 0, '>': 25, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, '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': '7344064', 'n_tokens_mistral': 1426, 'n_tokens_neox': 1338, 'n_words': 211}