Channel.h 6.7 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284
  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 -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. // Channel::Status
  119. OPENED,
  120. CLOSED,
  121. // SocketChannel::Status
  122. CONNECTING,
  123. CONNECTED,
  124. DISCONNECTED,
  125. } status;
  126. std::function<void(Buffer*)> onread;
  127. std::function<void(Buffer*)> onwrite;
  128. std::function<void()> onclose;
  129. private:
  130. static void on_read(hio_t* io, void* data, int readbytes) {
  131. Channel* channel = (Channel*)hio_context(io);
  132. if (channel && channel->onread) {
  133. Buffer buf(data, readbytes);
  134. channel->onread(&buf);
  135. }
  136. }
  137. static void on_write(hio_t* io, const void* data, int writebytes) {
  138. Channel* channel = (Channel*)hio_context(io);
  139. if (channel && channel->onwrite) {
  140. Buffer buf((void*)data, writebytes);
  141. channel->onwrite(&buf);
  142. }
  143. }
  144. static void on_close(hio_t* io) {
  145. Channel* channel = (Channel*)hio_context(io);
  146. if (channel) {
  147. channel->status = CLOSED;
  148. if (channel->onclose) {
  149. channel->onclose();
  150. }
  151. }
  152. }
  153. };
  154. class SocketChannel : public Channel {
  155. public:
  156. // for TcpClient
  157. std::function<void()> onconnect;
  158. std::function<void()> heartbeat;
  159. SocketChannel(hio_t* io) : Channel(io) {
  160. }
  161. virtual ~SocketChannel() {}
  162. int enableSSL() {
  163. return hio_enable_ssl(io_);
  164. }
  165. void setConnectTimeout(int timeout_ms) {
  166. if (io_ == NULL) return;
  167. hio_set_connect_timeout(io_, timeout_ms);
  168. }
  169. void setCloseTimeout(int timeout_ms) {
  170. if (io_ == NULL) return;
  171. hio_set_close_timeout(io_, timeout_ms);
  172. }
  173. void setKeepaliveTimeout(int timeout_ms) {
  174. if (io_ == NULL) return;
  175. hio_set_keepalive_timeout(io_, timeout_ms);
  176. }
  177. void setHeartbeat(int interval_ms, std::function<void()> fn) {
  178. if (io_ == NULL) return;
  179. heartbeat = std::move(fn);
  180. hio_set_heartbeat(io_, interval_ms, send_heartbeat);
  181. }
  182. void setUnpack(unpack_setting_t* setting) {
  183. if (io_ == NULL) return;
  184. hio_set_unpack(io_, setting);
  185. }
  186. int startConnect(int port, const char* host = "127.0.0.1") {
  187. sockaddr_u peeraddr;
  188. memset(&peeraddr, 0, sizeof(peeraddr));
  189. int ret = sockaddr_set_ipport(&peeraddr, host, port);
  190. if (ret != 0) {
  191. // hloge("unknown host %s", host);
  192. return ret;
  193. }
  194. return startConnect(&peeraddr.sa);
  195. }
  196. int startConnect(struct sockaddr* peeraddr) {
  197. if (io_ == NULL) return -1;
  198. hio_set_peeraddr(io_, peeraddr, SOCKADDR_LEN(peeraddr));
  199. return startConnect();
  200. }
  201. int startConnect() {
  202. if (io_ == NULL) return -1;
  203. status = CONNECTING;
  204. hio_setcb_connect(io_, on_connect);
  205. return hio_connect(io_);
  206. }
  207. bool isConnected() {
  208. return isOpened() && status == CONNECTED;
  209. }
  210. std::string localaddr() {
  211. if (io_ == NULL) return "";
  212. struct sockaddr* addr = hio_localaddr(io_);
  213. char buf[SOCKADDR_STRLEN] = {0};
  214. return SOCKADDR_STR(addr, buf);
  215. }
  216. std::string peeraddr() {
  217. if (io_ == NULL) return "";
  218. struct sockaddr* addr = hio_peeraddr(io_);
  219. char buf[SOCKADDR_STRLEN] = {0};
  220. return SOCKADDR_STR(addr, buf);
  221. }
  222. private:
  223. static void on_connect(hio_t* io) {
  224. SocketChannel* channel = (SocketChannel*)hio_context(io);
  225. if (channel) {
  226. channel->status = CONNECTED;
  227. if (channel->onconnect) {
  228. channel->onconnect();
  229. }
  230. }
  231. }
  232. static void send_heartbeat(hio_t* io) {
  233. SocketChannel* channel = (SocketChannel*)hio_context(io);
  234. if (channel && channel->heartbeat) {
  235. channel->heartbeat();
  236. }
  237. }
  238. };
  239. typedef std::shared_ptr<Channel> ChannelPtr;
  240. typedef std::shared_ptr<SocketChannel> SocketChannelPtr;
  241. }
  242. #endif // HV_CHANNEL_HPP_