RpcChannel.cc 2.2 KB

12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273747576777879808182838485
  1. #include <boost/asio.hpp>
  2. #include <boost/bind.hpp>
  3. #include <glog/logging.h>
  4. #include <google/protobuf/message.h>
  5. #include <google/protobuf/descriptor.h>
  6. #include "Rpc/RpcCommunicator.h"
  7. #include "Rpc/RpcChannel.h"
  8. #include "Rpc/RequestHandler.h"
  9. namespace Egametang {
  10. RpcChannel::RpcChannel(boost::asio::io_service& io_service, std::string host, int port):
  11. RpcCommunicator(io_service), id(0)
  12. {
  13. // another thread?
  14. boost::asio::ip::address address;
  15. address.from_string(host);
  16. boost::asio::ip::tcp::endpoint endpoint(address, port);
  17. socket.async_connect(endpoint,
  18. boost::bind(&RpcChannel::OnAsyncConnect, this,
  19. boost::asio::placeholders::error));
  20. }
  21. RpcChannel::~RpcChannel()
  22. {
  23. }
  24. void RpcChannel::OnAsyncConnect(const boost::system::error_code& err)
  25. {
  26. if (err)
  27. {
  28. LOG(ERROR) << "async connect failed: " << err.message();
  29. return;
  30. }
  31. RpcMetaPtr recv_meta(new RpcMeta());
  32. StringPtr recv_message(new std::string);
  33. RecvMeta(recv_meta, recv_message);
  34. }
  35. void RpcChannel::OnRecvMessage(RpcMetaPtr meta, StringPtr message)
  36. {
  37. RequestHandlerPtr request_handler = request_handlers[meta->id];
  38. request_handlers.erase(meta->id);
  39. request_handler->Response()->ParseFromString(*message);
  40. // meta和message可以循环利用
  41. RecvMeta(meta, message);
  42. // 回调放在函数最后.如果RecvMeta()放在回调之后,
  43. // 另外线程可能让io_service stop,导致RecvMeta还未跑完
  44. // 网络就终止了
  45. request_handler->Run();
  46. }
  47. void RpcChannel::OnSendMessage(RpcMetaPtr meta, StringPtr message)
  48. {
  49. }
  50. void RpcChannel::Stop()
  51. {
  52. socket.close();
  53. }
  54. void RpcChannel::CallMethod(
  55. const google::protobuf::MethodDescriptor* method,
  56. google::protobuf::RpcController* controller,
  57. const google::protobuf::Message* request,
  58. google::protobuf::Message* response,
  59. google::protobuf::Closure* done)
  60. {
  61. RequestHandlerPtr request_handler(new RequestHandler(response, done));
  62. request_handlers[++id] = request_handler;
  63. boost::hash<std::string> string_hash;
  64. StringPtr message(new std::string);
  65. request->SerializePartialToString(message.get());
  66. RpcMetaPtr meta(new RpcMeta());
  67. meta->size = message->size();
  68. meta->id = id;
  69. meta->method = string_hash(method->full_name());
  70. SendMeta(meta, message);
  71. }
  72. } // namespace Egametang