Channel.h 6.2 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263
  1. #ifndef HV_CHANNEL_HPP_
  2. #define HV_CHANNEL_HPP_
  3. #include <string>
  4. #include <functional>
  5. #include <memory>
  6. #include "hloop.h"
  7. #include "hsocket.h"
  8. #include "Buffer.h"
  9. namespace hv {
  10. class Channel {
  11. public:
  12. Channel(hio_t* io = NULL) {
  13. io_ = io;
  14. fd_ = -1;
  15. id_ = 0;
  16. ctx_ = NULL;
  17. if (io) {
  18. fd_ = hio_fd(io);
  19. id_ = hio_id(io);
  20. ctx_ = hio_context(io);
  21. hio_set_context(io, this);
  22. if (hio_getcb_read(io) == NULL) {
  23. hio_setcb_read(io_, on_read);
  24. }
  25. if (hio_getcb_write(io) == NULL) {
  26. hio_setcb_write(io_, on_write);
  27. }
  28. if (hio_getcb_close(io) == NULL) {
  29. hio_setcb_close(io_, on_close);
  30. }
  31. }
  32. status = isOpened() ? OPENED : CLOSED;
  33. }
  34. virtual ~Channel() {
  35. close();
  36. }
  37. hio_t* io() { return io_; }
  38. int fd() { return fd_; }
  39. uint32_t id() { return id_; }
  40. int error() { return hio_error(io_); }
  41. // context
  42. void* context() {
  43. return ctx_;
  44. }
  45. void setContext(void* ctx) {
  46. ctx_ = ctx;
  47. }
  48. template<class T>
  49. T* newContext() {
  50. ctx_ = new T;
  51. return (T*)ctx_;
  52. }
  53. template<class T>
  54. T* getContext() {
  55. return (T*)ctx_;
  56. }
  57. template<class T>
  58. void deleteContext() {
  59. if (ctx_) {
  60. delete (T*)ctx_;
  61. ctx_ = NULL;
  62. }
  63. }
  64. bool isOpened() {
  65. if (io_ == NULL) return false;
  66. return id_ == hio_id(io_) && hio_is_opened(io_);
  67. }
  68. bool isClosed() {
  69. return !isOpened();
  70. }
  71. int startRead() {
  72. if (!isOpened()) return 0;
  73. return hio_read_start(io_);
  74. }
  75. int stopRead() {
  76. if (!isOpened()) return 0;
  77. return hio_read_stop(io_);
  78. }
  79. int write(const void* data, int size) {
  80. if (!isOpened()) return 0;
  81. return hio_write(io_, data, size);
  82. }
  83. int write(Buffer* buf) {
  84. return write(buf->data(), buf->size());
  85. }
  86. int write(const std::string& str) {
  87. return write(str.data(), str.size());
  88. }
  89. int close(bool async = false) {
  90. if (!isOpened()) return 0;
  91. if (async) {
  92. return hio_close_async(io_);
  93. }
  94. return hio_close(io_);
  95. }
  96. public:
  97. hio_t* io_;
  98. int fd_;
  99. uint32_t id_;
  100. void* ctx_;
  101. enum Status {
  102. // Channel::Status
  103. OPENED,
  104. CLOSED,
  105. // SocketChannel::Status
  106. CONNECTING,
  107. CONNECTED,
  108. DISCONNECTED,
  109. } status;
  110. std::function<void(Buffer*)> onread;
  111. std::function<void(Buffer*)> onwrite;
  112. std::function<void()> onclose;
  113. private:
  114. static void on_read(hio_t* io, void* data, int readbytes) {
  115. Channel* channel = (Channel*)hio_context(io);
  116. if (channel && channel->onread) {
  117. Buffer buf(data, readbytes);
  118. channel->onread(&buf);
  119. }
  120. }
  121. static void on_write(hio_t* io, const void* data, int writebytes) {
  122. Channel* channel = (Channel*)hio_context(io);
  123. if (channel && channel->onwrite) {
  124. Buffer buf((void*)data, writebytes);
  125. channel->onwrite(&buf);
  126. }
  127. }
  128. static void on_close(hio_t* io) {
  129. Channel* channel = (Channel*)hio_context(io);
  130. if (channel) {
  131. channel->status = CLOSED;
  132. if (channel->onclose) {
  133. channel->onclose();
  134. }
  135. }
  136. }
  137. };
  138. class SocketChannel : public Channel {
  139. public:
  140. // for TcpClient
  141. std::function<void()> onconnect;
  142. std::function<void()> heartbeat;
  143. SocketChannel(hio_t* io) : Channel(io) {
  144. }
  145. virtual ~SocketChannel() {}
  146. int enableSSL() {
  147. return hio_enable_ssl(io_);
  148. }
  149. void setConnectTimeout(int timeout_ms) {
  150. if (io_ == NULL) return;
  151. hio_set_connect_timeout(io_, timeout_ms);
  152. }
  153. void setCloseTimeout(int timeout_ms) {
  154. if (io_ == NULL) return;
  155. hio_set_close_timeout(io_, timeout_ms);
  156. }
  157. void setKeepaliveTimeout(int timeout_ms) {
  158. if (io_ == NULL) return;
  159. hio_set_keepalive_timeout(io_, timeout_ms);
  160. }
  161. void setHeartbeat(int interval_ms, std::function<void()> fn) {
  162. if (io_ == NULL) return;
  163. heartbeat = std::move(fn);
  164. hio_set_heartbeat(io_, interval_ms, send_heartbeat);
  165. }
  166. int startConnect(int port, const char* host = "127.0.0.1") {
  167. sockaddr_u peeraddr;
  168. memset(&peeraddr, 0, sizeof(peeraddr));
  169. int ret = sockaddr_set_ipport(&peeraddr, host, port);
  170. if (ret != 0) {
  171. // hloge("unknown host %s", host);
  172. return ret;
  173. }
  174. return startConnect(&peeraddr.sa);
  175. }
  176. int startConnect(struct sockaddr* peeraddr) {
  177. if (io_ == NULL) return -1;
  178. hio_set_peeraddr(io_, peeraddr, SOCKADDR_LEN(peeraddr));
  179. return startConnect();
  180. }
  181. int startConnect() {
  182. if (io_ == NULL) return -1;
  183. status = CONNECTING;
  184. hio_setcb_connect(io_, on_connect);
  185. return hio_connect(io_);
  186. }
  187. bool isConnected() {
  188. return isOpened() && status == CONNECTED;
  189. }
  190. std::string localaddr() {
  191. if (io_ == NULL) return "";
  192. struct sockaddr* addr = hio_localaddr(io_);
  193. char buf[SOCKADDR_STRLEN] = {0};
  194. return SOCKADDR_STR(addr, buf);
  195. }
  196. std::string peeraddr() {
  197. if (io_ == NULL) return "";
  198. struct sockaddr* addr = hio_peeraddr(io_);
  199. char buf[SOCKADDR_STRLEN] = {0};
  200. return SOCKADDR_STR(addr, buf);
  201. }
  202. int send(const std::string& str) {
  203. return write(str);
  204. }
  205. private:
  206. static void on_connect(hio_t* io) {
  207. SocketChannel* channel = (SocketChannel*)hio_context(io);
  208. if (channel) {
  209. channel->status = CONNECTED;
  210. if (channel->onconnect) {
  211. channel->onconnect();
  212. }
  213. }
  214. }
  215. static void send_heartbeat(hio_t* io) {
  216. SocketChannel* channel = (SocketChannel*)hio_context(io);
  217. if (channel && channel->heartbeat) {
  218. channel->heartbeat();
  219. }
  220. }
  221. };
  222. typedef std::shared_ptr<Channel> ChannelPtr;
  223. typedef std::shared_ptr<SocketChannel> SocketChannelPtr;
  224. }
  225. #endif // HV_CHANNEL_HPP_