Channel.h 6.9 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291
  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 || status >= DISCONNECTED) 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 -1;
  73. return hio_read_start(io_);
  74. }
  75. int stopRead() {
  76. if (!isOpened()) return -1;
  77. return hio_read_stop(io_);
  78. }
  79. int readOnce() {
  80. if (!isOpened()) return -1;
  81. return hio_read_once(io_);
  82. }
  83. int readString() {
  84. if (!isOpened()) return -1;
  85. return hio_readstring(io_);
  86. }
  87. int readLine() {
  88. if (!isOpened()) return -1;
  89. return hio_readline(io_);
  90. }
  91. int readBytes(int len) {
  92. if (!isOpened() || len <= 0) return -1;
  93. return hio_readbytes(io_, len);
  94. }
  95. int write(const void* data, int size) {
  96. if (!isOpened()) return -1;
  97. return hio_write(io_, data, size);
  98. }
  99. int write(Buffer* buf) {
  100. return write(buf->data(), buf->size());
  101. }
  102. int write(const std::string& str) {
  103. return write(str.data(), str.size());
  104. }
  105. int close(bool async = false) {
  106. if (!isOpened()) return -1;
  107. if (async) {
  108. return hio_close_async(io_);
  109. }
  110. return hio_close(io_);
  111. }
  112. public:
  113. hio_t* io_;
  114. int fd_;
  115. uint32_t id_;
  116. void* ctx_;
  117. enum Status {
  118. OPENED,
  119. CONNECTING,
  120. CONNECTED,
  121. DISCONNECTED,
  122. CLOSED,
  123. } status;
  124. std::function<void(Buffer*)> onread;
  125. std::function<void(Buffer*)> onwrite;
  126. std::function<void()> onclose;
  127. private:
  128. static void on_read(hio_t* io, void* data, int readbytes) {
  129. Channel* channel = (Channel*)hio_context(io);
  130. if (channel && channel->onread) {
  131. Buffer buf(data, readbytes);
  132. channel->onread(&buf);
  133. }
  134. }
  135. static void on_write(hio_t* io, const void* data, int writebytes) {
  136. Channel* channel = (Channel*)hio_context(io);
  137. if (channel && channel->onwrite) {
  138. Buffer buf((void*)data, writebytes);
  139. channel->onwrite(&buf);
  140. }
  141. }
  142. static void on_close(hio_t* io) {
  143. Channel* channel = (Channel*)hio_context(io);
  144. if (channel) {
  145. channel->status = CLOSED;
  146. if (channel->onclose) {
  147. channel->onclose();
  148. }
  149. }
  150. }
  151. };
  152. class SocketChannel : public Channel {
  153. public:
  154. std::function<void()> onconnect; // only for TcpClient
  155. std::function<void()> heartbeat;
  156. SocketChannel(hio_t* io) : Channel(io) {
  157. }
  158. virtual ~SocketChannel() {}
  159. int enableSSL() {
  160. return hio_enable_ssl(io_);
  161. }
  162. void setConnectTimeout(int timeout_ms) {
  163. if (io_ == NULL) return;
  164. hio_set_connect_timeout(io_, timeout_ms);
  165. }
  166. void setCloseTimeout(int timeout_ms) {
  167. if (io_ == NULL) return;
  168. hio_set_close_timeout(io_, timeout_ms);
  169. }
  170. void setReadTimeout(int timeout_ms) {
  171. if (io_ == NULL) return;
  172. hio_set_read_timeout(io_, timeout_ms);
  173. }
  174. void setWriteTimeout(int timeout_ms) {
  175. if (io_ == NULL) return;
  176. hio_set_write_timeout(io_, timeout_ms);
  177. }
  178. void setKeepaliveTimeout(int timeout_ms) {
  179. if (io_ == NULL) return;
  180. hio_set_keepalive_timeout(io_, timeout_ms);
  181. }
  182. void setHeartbeat(int interval_ms, std::function<void()> fn) {
  183. if (io_ == NULL) return;
  184. heartbeat = std::move(fn);
  185. hio_set_heartbeat(io_, interval_ms, send_heartbeat);
  186. }
  187. void setUnpack(unpack_setting_t* setting) {
  188. if (io_ == NULL) return;
  189. hio_set_unpack(io_, setting);
  190. }
  191. int startConnect(int port, const char* host = "127.0.0.1") {
  192. sockaddr_u peeraddr;
  193. memset(&peeraddr, 0, sizeof(peeraddr));
  194. int ret = sockaddr_set_ipport(&peeraddr, host, port);
  195. if (ret != 0) {
  196. // hloge("unknown host %s", host);
  197. return ret;
  198. }
  199. return startConnect(&peeraddr.sa);
  200. }
  201. int startConnect(struct sockaddr* peeraddr) {
  202. if (io_ == NULL) return -1;
  203. hio_set_peeraddr(io_, peeraddr, SOCKADDR_LEN(peeraddr));
  204. return startConnect();
  205. }
  206. int startConnect() {
  207. if (io_ == NULL) return -1;
  208. status = CONNECTING;
  209. hio_setcb_connect(io_, on_connect);
  210. return hio_connect(io_);
  211. }
  212. bool isConnected() {
  213. return status == CONNECTED && isOpened();
  214. }
  215. std::string localaddr() {
  216. if (io_ == NULL) return "";
  217. struct sockaddr* addr = hio_localaddr(io_);
  218. char buf[SOCKADDR_STRLEN] = {0};
  219. return SOCKADDR_STR(addr, buf);
  220. }
  221. std::string peeraddr() {
  222. if (io_ == NULL) return "";
  223. struct sockaddr* addr = hio_peeraddr(io_);
  224. char buf[SOCKADDR_STRLEN] = {0};
  225. return SOCKADDR_STR(addr, buf);
  226. }
  227. private:
  228. static void on_connect(hio_t* io) {
  229. SocketChannel* channel = (SocketChannel*)hio_context(io);
  230. if (channel) {
  231. channel->status = CONNECTED;
  232. if (channel->onconnect) {
  233. channel->onconnect();
  234. }
  235. }
  236. }
  237. static void send_heartbeat(hio_t* io) {
  238. SocketChannel* channel = (SocketChannel*)hio_context(io);
  239. if (channel && channel->heartbeat) {
  240. channel->heartbeat();
  241. }
  242. }
  243. };
  244. typedef std::shared_ptr<Channel> ChannelPtr;
  245. typedef std::shared_ptr<SocketChannel> SocketChannelPtr;
  246. }
  247. #endif // HV_CHANNEL_HPP_