with extn control for each service

Author: sroycode

CMakeLists.txt

@@ -1,8 +1,8 @@
 set(CMAKE_MODULE_PATH ${PROJECT_SOURCE_DIR}/cmake_modules)
 
+set(CMAKE_BUILD_TYPE "Release")
 if (NOT CMAKE_BUILD_TYPE)
   set(CMAKE_BUILD_TYPE "Debug")
-  #set(CMAKE_BUILD_TYPE "Release")
 endif()
 message("-- Build type: ${CMAKE_BUILD_TYPE}")
 

include/zqrpc/RpcServer.hh

@@ -64,6 +64,7 @@ public:
 protected:
 
 	zmq::context_t* context_;
+	const char* suffix_;
 	const std::string use_inproc_workil;
 	const std::string use_inproc_worker;
 	const std::string use_inproc_pcontrol;

src/RpcServer.cc

@@ -40,9 +40,10 @@
 
 zqrpc::RpcServer::RpcServer(zmq::context_t* context,const char* suffix) :
 	context_(context),
-	use_inproc_workil( std::string(ZQRPC_INPROC_WORKIL)+std::string(suffix) ),
-	use_inproc_worker( std::string(ZQRPC_INPROC_WORKER)+std::string(suffix) ),
-	use_inproc_pcontrol( std::string(ZQRPC_INPROC_PCONTROL)+std::string(suffix) ),
+	suffix_(suffix),
+	use_inproc_workil( std::string(ZQRPC_INPROC_WORKIL)+std::string(suffix_) ),
+	use_inproc_worker( std::string(ZQRPC_INPROC_WORKER)+std::string(suffix_) ),
+	use_inproc_pcontrol( std::string(ZQRPC_INPROC_PCONTROL)+std::string(suffix_) ),
 	rpc_frontend_(context_,ZMQ_ROUTER,"ROUTER"),
 	rpc_backend_(context_,ZMQ_DEALER,"DEALER"),
 	rpc_control_(context_,ZMQ_DEALER,"CONTROL"),
@@ -95,11 +96,15 @@ void zqrpc::RpcServer::Start(std::size_t noof_threads)
 			rpc_frontend_.bind(it->c_str());
 		}
 		rpc_frontend_.bind(use_inproc_workil.c_str());
+		DLOG(INFO) << "Bound " << use_inproc_workil.c_str() << std::endl;
 		rpc_backend_.bind(use_inproc_worker.c_str());
+		DLOG(INFO) << "Bound " << use_inproc_worker.c_str() << std::endl;
 		rpc_control_.bind(use_inproc_pcontrol.c_str());
+		DLOG(INFO) << "Bound " << use_inproc_pcontrol.c_str() << std::endl;
+		
 		started_=true;
 		for(std::size_t i = 0; i < noof_threads; ++i) {
-			threads_.push_back( new boost::thread(zqrpc::RpcWorker(context_, rpc_method_map_)) );
+			threads_.push_back( new boost::thread(zqrpc::RpcWorker(context_, rpc_method_map_,suffix_)) );
 		}
 		ProxyStart();
 		DLOG(INFO) << "Loop Ended " << std::endl;

src/RpcWorker.cc

@@ -51,6 +51,7 @@ zqrpc::RpcWorker::~RpcWorker() {}
 void zqrpc::RpcWorker::operator() ()
 {
 	zqrpc::ZSocket socket(context_,ZMQ_DEALER,boost::lexical_cast<std::string>(boost::this_thread::get_id()));
+	DLOG(INFO) << "Connecting " << use_inproc_worker.c_str() << std::endl;
 	socket.connect(use_inproc_worker.c_str());
 	socket.SetLinger(0);
 	google::protobuf::Message* request = NULL;
@@ -113,7 +114,9 @@ void zqrpc::RpcWorker::operator() ()
 	}
 	delete request;
 	delete response;
+	DLOG(INFO) << "Disconnecting " << use_inproc_worker.c_str() << std::endl;
 	socket.disconnect(use_inproc_worker.c_str());
+	DLOG(INFO) << "Disconnected " << use_inproc_worker.c_str() << std::endl;
 	socket.close();
 	/**
 	for (RpcMethodMapT::iterator it = rpc_method_map_.begin(); it != rpc_method_map_.end();) {