RpcCommunicator.cc 2.6 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110
  1. #include <boost/bind.hpp>
  2. #include <boost/asio.hpp>
  3. #include <boost/lexical_cast.hpp>
  4. #include <glog/logging.h>
  5. #include "Rpc/RpcCommunicator.h"
  6. namespace Egametang {
  7. RpcCommunicator::RpcCommunicator(boost::asio::io_service& service):
  8. io_service(service), socket(service)
  9. {
  10. }
  11. RpcCommunicator::~RpcCommunicator()
  12. {
  13. }
  14. boost::asio::ip::tcp::socket& RpcCommunicator::Socket()
  15. {
  16. return socket;
  17. }
  18. void RpcCommunicator::Stop()
  19. {
  20. socket.close();
  21. }
  22. void RpcCommunicator::RecvMeta(RpcMetaPtr meta, StringPtr message)
  23. {
  24. boost::asio::async_read(socket,
  25. boost::asio::buffer(reinterpret_cast<char*>(meta.get()), sizeof(*meta)),
  26. boost::bind(&RpcCommunicator::RecvMessage, this,
  27. meta, message, boost::asio::placeholders::error));
  28. }
  29. void RpcCommunicator::RecvMessage(RpcMetaPtr meta, StringPtr message,
  30. const boost::system::error_code& err)
  31. {
  32. if (err)
  33. {
  34. LOG(ERROR) << "receive message size failed: " << err.message();
  35. VLOG(2) << "meta: " << meta->ToString() << " message: " << *message;
  36. Stop();
  37. return;
  38. }
  39. message->resize(meta->size, '\0');
  40. boost::asio::async_read(socket,
  41. boost::asio::buffer(reinterpret_cast<char*>(&message->at(0)), meta->size),
  42. boost::bind(&RpcCommunicator::RecvDone, this,
  43. meta, message, boost::asio::placeholders::error));
  44. }
  45. void RpcCommunicator::RecvDone(RpcMetaPtr meta, StringPtr message,
  46. const boost::system::error_code& err)
  47. {
  48. if (err)
  49. {
  50. LOG(ERROR) << "receive message failed: " << err.message();
  51. Stop();
  52. return;
  53. }
  54. OnRecvMessage(meta, message);
  55. }
  56. void RpcCommunicator::OnRecvMessage(RpcMetaPtr meta, StringPtr message)
  57. {
  58. }
  59. void RpcCommunicator::SendMeta(RpcMetaPtr meta, StringPtr message)
  60. {
  61. CHECK_EQ(meta->size, message->size()) << "meta and message size not match!";
  62. boost::asio::async_write(socket,
  63. boost::asio::buffer(reinterpret_cast<char*>(meta.get()), sizeof(*meta)),
  64. boost::bind(&RpcCommunicator::SendMessage, this,
  65. meta, message, boost::asio::placeholders::error));
  66. }
  67. void RpcCommunicator::SendMessage(RpcMetaPtr meta, StringPtr message,
  68. const boost::system::error_code& err)
  69. {
  70. if (err)
  71. {
  72. LOG(ERROR) << "send message size failed: " << err.message();
  73. Stop();
  74. return;
  75. }
  76. boost::asio::async_write(socket, boost::asio::buffer(*message),
  77. boost::bind(&RpcCommunicator::SendDone, this,
  78. meta, message, boost::asio::placeholders::error));
  79. }
  80. void RpcCommunicator::SendDone(RpcMetaPtr meta, StringPtr message,
  81. const boost::system::error_code& err)
  82. {
  83. if (err)
  84. {
  85. LOG(ERROR) << "send message failed: " << err.message();
  86. Stop();
  87. return;
  88. }
  89. OnSendMessage(meta, message);
  90. }
  91. void RpcCommunicator::OnSendMessage(RpcMetaPtr meta, StringPtr message)
  92. {
  93. }
  94. } // namespace Egametang