protorpc_server.cpp 4.0 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130
  1. /*
  2. * proto rpc server
  3. *
  4. * @build make protorpc
  5. * @server bin/protorpc_server 1234
  6. * @client bin/protorpc_client 127.0.0.1 1234 add 1 2
  7. *
  8. */
  9. #include "hloop.h"
  10. #include "hbase.h"
  11. #include "hsocket.h"
  12. #include "TcpServer.h"
  13. using namespace hv;
  14. #include "protorpc.h"
  15. #include "router.h"
  16. #include "handler/handler.h"
  17. #include "handler/calc.h"
  18. #include "handler/login.h"
  19. protorpc_router router[] = {
  20. {"add", calc_add},
  21. {"sub", calc_sub},
  22. {"mul", calc_mul},
  23. {"div", calc_div},
  24. {"login", login},
  25. };
  26. #define PROTORPC_ROUTER_NUM (sizeof(router)/sizeof(router[0]))
  27. class ProtoRpcServer : public TcpServer {
  28. public:
  29. ProtoRpcServer() : TcpServer()
  30. {
  31. onConnection = [](const SocketChannelPtr& channel) {
  32. std::string peeraddr = channel->peeraddr();
  33. if (channel->isConnected()) {
  34. printf("%s connected! connfd=%d\n", peeraddr.c_str(), channel->fd());
  35. } else {
  36. printf("%s disconnected! connfd=%d\n", peeraddr.c_str(), channel->fd());
  37. }
  38. };
  39. onMessage = handleMessage;
  40. // init protorpc_unpack_setting
  41. unpack_setting_t protorpc_unpack_setting;
  42. memset(&protorpc_unpack_setting, 0, sizeof(unpack_setting_t));
  43. protorpc_unpack_setting.mode = UNPACK_BY_LENGTH_FIELD;
  44. protorpc_unpack_setting.package_max_length = DEFAULT_PACKAGE_MAX_LENGTH;
  45. protorpc_unpack_setting.body_offset = PROTORPC_HEAD_LENGTH;
  46. protorpc_unpack_setting.length_field_offset = 1;
  47. protorpc_unpack_setting.length_field_bytes = 4;
  48. protorpc_unpack_setting.length_field_coding = ENCODE_BY_BIG_ENDIAN;
  49. setUnpack(&protorpc_unpack_setting);
  50. }
  51. int listen(int port) { return createsocket(port); }
  52. private:
  53. static void handleMessage(const SocketChannelPtr& channel, Buffer* buf) {
  54. // unpack -> Request::ParseFromArray -> router -> Response::SerializeToArray -> pack -> Channel::write
  55. // protorpc_unpack
  56. protorpc_message msg;
  57. memset(&msg, 0, sizeof(msg));
  58. int packlen = protorpc_unpack(&msg, buf->data(), buf->size());
  59. if (packlen < 0) {
  60. printf("protorpc_unpack failed!\n");
  61. return;
  62. }
  63. assert(packlen == buf->size());
  64. // Request::ParseFromArray
  65. protorpc::Request req;
  66. protorpc::Response res;
  67. if (req.ParseFromArray(msg.body, msg.head.length)) {
  68. printf("> %s\n", req.DebugString().c_str());
  69. res.set_id(req.id());
  70. // router
  71. const char* method = req.method().c_str();
  72. bool found = false;
  73. for (int i = 0; i < PROTORPC_ROUTER_NUM; ++i) {
  74. if (strcmp(method, router[i].method) == 0) {
  75. found = true;
  76. router[i].handler(req, &res);
  77. break;
  78. }
  79. }
  80. if (!found) {
  81. not_found(req, &res);
  82. }
  83. } else {
  84. bad_request(req, &res);
  85. }
  86. // Response::SerializeToArray + protorpc_pack
  87. memset(&msg, 0, sizeof(msg));
  88. msg.head.length = res.ByteSizeLong();
  89. packlen = protorpc_package_length(&msg.head);
  90. unsigned char* writebuf = NULL;
  91. HV_ALLOC(writebuf, packlen);
  92. packlen = protorpc_pack(&msg, writebuf, packlen);
  93. if (packlen > 0) {
  94. printf("< %s\n", res.DebugString().c_str());
  95. res.SerializeToArray(writebuf + PROTORPC_HEAD_LENGTH, msg.head.length);
  96. channel->write(writebuf, packlen);
  97. }
  98. HV_FREE(writebuf);
  99. }
  100. };
  101. int main(int argc, char** argv) {
  102. if (argc < 2) {
  103. printf("Usage: %s port\n", argv[0]);
  104. return -10;
  105. }
  106. int port = atoi(argv[1]);
  107. ProtoRpcServer srv;
  108. int listenfd = srv.listen(port);
  109. if (listenfd < 0) {
  110. return -20;
  111. }
  112. printf("protorpc_server listen on port %d, listenfd=%d ...\n", port, listenfd);
  113. srv.setThreadNum(4);
  114. srv.start();
  115. while (1) hv_sleep(1);
  116. return 0;
  117. }