远程过程调用-buttonrpc源码解析8-ZMQ网络库

buttonrpc中，通过ZMQ的方式进行网络调用
ZMQ(Zero Message Queue)是一种消息队列，核心引擎由C++编写，是轻量级消息通信库，是在对传统的标准Socket接口扩展的基础上形成的特色消息通信中间件。ZMQ有多种模式可以使用，常用的模式包括Request-Reply(请求-应答)，Publish/Subscribe（发布/订阅）和Push/Pull（推/拉）三种模式。
1、buttonrpc中网络相关代码：Request-Reply(请求-应答)

// 省略其它实现，只展示网络部分...
class buttonrpc
{
public:
	// network
	void as_client(std::string ip, int port);
	void as_server(int port);
	void send(zmq::message_t& data);
	void recv(zmq::message_t& data);
	void set_timeout(uint32_t ms);
	void run();
private:
	std::map<std::string, std::function<void(Serializer*, const char*, int)>> m_handlers;

	zmq::context_t m_context;
	std::unique_ptr<zmq::socket_t, std::function<void(zmq::socket_t*)>> m_socket;

	rpc_err_code m_error_code;

	int m_role;
}

// network
inline void buttonrpc::as_client( std::string ip, int port )
{
	m_role = RPC_CLIENT;
	m_socket = std::unique_ptr<zmq::socket_t, std::function<void(zmq::socket_t*)>>(new zmq::socket_t(m_context, ZMQ_REQ), [](zmq::socket_t* sock){ sock->close(); delete sock; sock =nullptr;});
	ostringstream os;
	os << "tcp://" << ip << ":" << port;
	m_socket->connect (os.str());
}

inline void buttonrpc::as_server( int port )
{
	m_role = RPC_SERVER;
	m_socket = std::unique_ptr<zmq::socket_t, std::function<void(zmq::socket_t*)>>(new zmq::socket_t(m_context, ZMQ_REP), [](zmq::socket_t* sock){ sock->close(); delete sock; sock =nullptr;});
	ostringstream os;
	os << "tcp://*:" << port;
	m_socket->bind (os.str());
}

inline void buttonrpc::send( zmq::message_t& data )
{
	m_socket->send(data);
}

inline void buttonrpc::recv( zmq::message_t& data )
{
	m_socket->recv(&data);
}

inline void buttonrpc::set_timeout(uint32_t ms)
{
	// only client can set
	if (m_role == RPC_CLIENT) {
		m_socket->setsockopt(ZMQ_RCVTIMEO, ms);
	}
}

inline void buttonrpc::run()
{
	// only server can call
	if (m_role != RPC_SERVER) {
		return;
	}
	while (1){
		zmq::message_t data;
		recv(data);
		StreamBuffer iodev((char*)data.data(), data.size());
		Serializer ds(iodev);

		std::string funname;
		ds >> funname;
		Serializer* r = call_(funname, ds.current(), ds.size()- funname.size());

		zmq::message_t retmsg (r->size());
		memcpy (retmsg.data (), r->data(), r->size());
		send(retmsg);
		delete r;
	}
}

2、服务端

// 以此函数为例
int foo_3(int arg1) {
	buttont_assert(arg1 == 10);
	return arg1 * arg1;
}

int main()
{
	buttonrpc server;
	server.as_server(5555);

	server.bind("foo_3", std::function<int(int)>(foo_3));
	std::cout << "run rpc server on: " << 5555 << std::endl;
	server.run();

	return 0;
}

温馨提示：server.bind("foo_3", std::function<int(int)>(foo_3));与server.bind("foo_3", foo_3);都可以
3、客户端

int main()
{
	buttonrpc client;
	client.as_client("127.0.0.1", 5555);
	client.set_timeout(2000);

	int foo3r = client.call<int>("foo_3", 10).val();
	buttont_assert(foo3r == 100);

	return 0;
}

温馨提示：buttont_assert(foo3r == 100);表示断言，用来判断返回值foo3r==100条件是否为真，源码如下：

#define buttont_assert(exp) { \
	if (!(exp)) {\
		std::cout << "ERROR: "; \
		std::cout << "function: " << __FUNCTION__  << ", line: " <<  __LINE__ << std::endl; \
		system("pause"); \
	}\
}\