Channel.h 7.0 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294
  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. status = CLOSED;
  18. if (io) {
  19. fd_ = hio_fd(io);
  20. id_ = hio_id(io);
  21. ctx_ = hio_context(io);
  22. hio_set_context(io, this);
  23. if (hio_is_opened(io)) {
  24. status = OPENED;
  25. }
  26. if (hio_getcb_read(io) == NULL) {
  27. hio_setcb_read(io_, on_read);
  28. }
  29. if (hio_getcb_write(io) == NULL) {
  30. hio_setcb_write(io_, on_write);
  31. }
  32. if (hio_getcb_close(io) == NULL) {
  33. hio_setcb_close(io_, on_close);
  34. }
  35. }
  36. }
  37. virtual ~Channel() {
  38. close();
  39. }
  40. hio_t* io() { return io_; }
  41. int fd() { return fd_; }
  42. uint32_t id() { return id_; }
  43. int error() { return hio_error(io_); }
  44. // context
  45. void* context() {
  46. return ctx_;
  47. }
  48. void setContext(void* ctx) {
  49. ctx_ = ctx;
  50. }
  51. template<class T>
  52. T* newContext() {
  53. ctx_ = new T;
  54. return (T*)ctx_;
  55. }
  56. template<class T>
  57. T* getContext() {
  58. return (T*)ctx_;
  59. }
  60. template<class T>
  61. void deleteContext() {
  62. if (ctx_) {
  63. delete (T*)ctx_;
  64. ctx_ = NULL;
  65. }
  66. }
  67. bool isOpened() {
  68. if (io_ == NULL || status >= DISCONNECTED) return false;
  69. return id_ == hio_id(io_) && hio_is_opened(io_);
  70. }
  71. bool isClosed() {
  72. return !isOpened();
  73. }
  74. int startRead() {
  75. if (!isOpened()) return -1;
  76. return hio_read_start(io_);
  77. }
  78. int stopRead() {
  79. if (!isOpened()) return -1;
  80. return hio_read_stop(io_);
  81. }
  82. int readOnce() {
  83. if (!isOpened()) return -1;
  84. return hio_read_once(io_);
  85. }
  86. int readString() {
  87. if (!isOpened()) return -1;
  88. return hio_readstring(io_);
  89. }
  90. int readLine() {
  91. if (!isOpened()) return -1;
  92. return hio_readline(io_);
  93. }
  94. int readBytes(int len) {
  95. if (!isOpened() || len <= 0) return -1;
  96. return hio_readbytes(io_, len);
  97. }
  98. int write(const void* data, int size) {
  99. if (!isOpened()) return -1;
  100. return hio_write(io_, data, size);
  101. }
  102. int write(Buffer* buf) {
  103. return write(buf->data(), buf->size());
  104. }
  105. int write(const std::string& str) {
  106. return write(str.data(), str.size());
  107. }
  108. int close(bool async = false) {
  109. if (!isOpened()) return -1;
  110. if (async) {
  111. return hio_close_async(io_);
  112. }
  113. return hio_close(io_);
  114. }
  115. public:
  116. hio_t* io_;
  117. int fd_;
  118. uint32_t id_;
  119. void* ctx_;
  120. enum Status {
  121. OPENED,
  122. CONNECTING,
  123. CONNECTED,
  124. DISCONNECTED,
  125. CLOSED,
  126. } status;
  127. std::function<void(Buffer*)> onread;
  128. std::function<void(Buffer*)> onwrite;
  129. std::function<void()> onclose;
  130. private:
  131. static void on_read(hio_t* io, void* data, int readbytes) {
  132. Channel* channel = (Channel*)hio_context(io);
  133. if (channel && channel->onread) {
  134. Buffer buf(data, readbytes);
  135. channel->onread(&buf);
  136. }
  137. }
  138. static void on_write(hio_t* io, const void* data, int writebytes) {
  139. Channel* channel = (Channel*)hio_context(io);
  140. if (channel && channel->onwrite) {
  141. Buffer buf((void*)data, writebytes);
  142. channel->onwrite(&buf);
  143. }
  144. }
  145. static void on_close(hio_t* io) {
  146. Channel* channel = (Channel*)hio_context(io);
  147. if (channel) {
  148. channel->status = CLOSED;
  149. if (channel->onclose) {
  150. channel->onclose();
  151. }
  152. }
  153. }
  154. };
  155. class SocketChannel : public Channel {
  156. public:
  157. std::function<void()> onconnect; // only for TcpClient
  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 setReadTimeout(int timeout_ms) {
  174. if (io_ == NULL) return;
  175. hio_set_read_timeout(io_, timeout_ms);
  176. }
  177. void setWriteTimeout(int timeout_ms) {
  178. if (io_ == NULL) return;
  179. hio_set_write_timeout(io_, timeout_ms);
  180. }
  181. void setKeepaliveTimeout(int timeout_ms) {
  182. if (io_ == NULL) return;
  183. hio_set_keepalive_timeout(io_, timeout_ms);
  184. }
  185. void setHeartbeat(int interval_ms, std::function<void()> fn) {
  186. if (io_ == NULL) return;
  187. heartbeat = std::move(fn);
  188. hio_set_heartbeat(io_, interval_ms, send_heartbeat);
  189. }
  190. void setUnpack(unpack_setting_t* setting) {
  191. if (io_ == NULL) return;
  192. hio_set_unpack(io_, setting);
  193. }
  194. int startConnect(int port, const char* host = "127.0.0.1") {
  195. sockaddr_u peeraddr;
  196. memset(&peeraddr, 0, sizeof(peeraddr));
  197. int ret = sockaddr_set_ipport(&peeraddr, host, port);
  198. if (ret != 0) {
  199. // hloge("unknown host %s", host);
  200. return ret;
  201. }
  202. return startConnect(&peeraddr.sa);
  203. }
  204. int startConnect(struct sockaddr* peeraddr) {
  205. if (io_ == NULL) return -1;
  206. hio_set_peeraddr(io_, peeraddr, SOCKADDR_LEN(peeraddr));
  207. return startConnect();
  208. }
  209. int startConnect() {
  210. if (io_ == NULL) return -1;
  211. status = CONNECTING;
  212. hio_setcb_connect(io_, on_connect);
  213. return hio_connect(io_);
  214. }
  215. bool isConnected() {
  216. return status == CONNECTED && isOpened();
  217. }
  218. std::string localaddr() {
  219. if (io_ == NULL) return "";
  220. struct sockaddr* addr = hio_localaddr(io_);
  221. char buf[SOCKADDR_STRLEN] = {0};
  222. return SOCKADDR_STR(addr, buf);
  223. }
  224. std::string peeraddr() {
  225. if (io_ == NULL) return "";
  226. struct sockaddr* addr = hio_peeraddr(io_);
  227. char buf[SOCKADDR_STRLEN] = {0};
  228. return SOCKADDR_STR(addr, buf);
  229. }
  230. private:
  231. static void on_connect(hio_t* io) {
  232. SocketChannel* channel = (SocketChannel*)hio_context(io);
  233. if (channel) {
  234. channel->status = CONNECTED;
  235. if (channel->onconnect) {
  236. channel->onconnect();
  237. }
  238. }
  239. }
  240. static void send_heartbeat(hio_t* io) {
  241. SocketChannel* channel = (SocketChannel*)hio_context(io);
  242. if (channel && channel->heartbeat) {
  243. channel->heartbeat();
  244. }
  245. }
  246. };
  247. typedef std::shared_ptr<Channel> ChannelPtr;
  248. typedef std::shared_ptr<SocketChannel> SocketChannelPtr;
  249. }
  250. #endif // HV_CHANNEL_HPP_