text: /*
   +----------------------------------------------------------------------+
   | Swoole                                                               |
   +----------------------------------------------------------------------+
   | This source file is subject to version 2.0 of the Apache license,    |
   | that is bundled with this package in the file LICENSE, and is        |
   | available through the world-wide-web at the following url:           |
   | http://www.apache.org/licenses/LICENSE-2.0.html                      |
   | If you did not receive a copy of the Apache2.0 license and are unable|
   | to obtain it through the world-wide-web, please send a note to       |
   | license@swoole.com so we can mail you a copy immediately.            |
   +----------------------------------------------------------------------+
   | Author: Tianfeng Han  <mikan.tenny@gmail.com>                        |
   +----------------------------------------------------------------------+
   */
  
  #include "swoole.h"
  #include "swoole_api.h"
  #include "swoole_string.h"
  #include "swoole_socket.h"
  #include "swoole_reactor.h"
  #include "swoole_timer.h"
  #include "swoole_protocol.h"
  #include "swoole_client.h"
  #include "swoole_proxy.h"
  #include "swoole_async.h"
  
  #include <assert.h>
  
  namespace swoole {
  namespace network {
  
  static int Client_inet_addr(Client *cli, const char *host, int port);
  static int Client_tcp_connect_sync(Client *cli, const char *host, int port, double _timeout, int udp_connect);
  static int Client_tcp_connect_async(Client *cli, const char *host, int port, double timeout, int nonblock);
  static int Client_udp_connect(Client *cli, const char *host, int port, double _timeout, int udp_connect);
  
  static ssize_t Client_tcp_send_sync(Client *cli, const char *data, size_t length, int flags);
  static ssize_t Client_tcp_send_async(Client *cli, const char *data, size_t length, int flags);
  static ssize_t Client_udp_send(Client *cli, const char *data, size_t length, int flags);
  
  static int Client_tcp_sendfile_sync(Client *cli, const char *filename, off_t offset, size_t length);
  static int Client_tcp_sendfile_async(Client *cli, const char *filename, off_t offset, size_t length);
  
  static ssize_t Client_tcp_recv_no_buffer(Client *cli, char *data, size_t len, int flags);
  static ssize_t Client_udp_recv(Client *cli, char *data, size_t len, int waitall);
  
  static int Client_onDgramRead(Reactor *reactor, Event *event);
  static int Client_onStreamRead(Reactor *reactor, Event *event);
  static int Client_onWrite(Reactor *reactor, Event *event);
  static int Client_onError(Reactor *reactor, Event *event);
  static void Client_onTimeout(Timer *timer, TimerNode *tnode);
  static void Client_onResolveCompleted(AsyncEvent *event);
  static int Client_onPackage(Protocol *proto, Socket *conn, const char *data, uint32_t length);
  
  static sw_inline void execute_onConnect(Client *cli) {
      if (cli->timer) {
          swoole_timer_del(cli->timer);
          cli->timer = nullptr;
      }
      cli->onConnect(cli);
  }
  
  void Client::init_reactor(Reactor *reactor) {
      reactor->set_handler(SW_FD_STREAM_CLIENT | SW_EVENT_READ, Client_onStreamRead);
      reactor->set_handler(SW_FD_DGRAM_CLIENT | SW_EVENT_READ, Client_onDgramRead);
      reactor->set_handler(SW_FD_STREAM_CLIENT | SW_EVENT_WRITE, Client_onWrite);
      reactor->set_handler(SW_FD_STREAM_CLIENT | SW_EVENT_ERROR, Client_onError);
  }
  
  Client::Client(enum swSocket_type _type, bool _async) : async(_async) {
      fd_type = Socket::is_stream(_type) ? SW_FD_STREAM_CLIENT : SW_FD_DGRAM_CLIENT;
      socket = swoole::make_socket(_type, fd_type, (async ? SW_SOCK_NONBLOCK : 0) | SW_SOCK_CLOEXEC);
      if (socket == nullptr) {
          swSysWarn("socket() failed");
          return;
      }
  
      socket->object = this;
      input_buffer_size = SW_CLIENT_BUFFER_SIZE;
      socket->chunk_size = SW_SEND_BUFFER_SIZE;
  
      if (socket->is_stream()) {
          recv = Client_tcp_recv_no_buffer;
          if (async) {
              connect = Client_tcp_connect_async;
              send = Client_tcp_send_async;
              sendfile = Client_tcp_sendfile_async;
              socket->dontwait = SwooleG.socket_dontwait;
          } else {
              connect = Client_tcp_connect_sync;
              send = Client_tcp_send_sync;
              sendfile = Client_tcp_sendfile_sync;
          }
      } else {
          connect = Client_udp_connect;
          recv = Client_udp_recv;
          send = Client_udp_send;
      }
  
      Socket::get_domain_and_type(_type, &_sock_domain, &_sock_type);
  
      protocol.package_length_type = 'N';
      protocol.package_length_size = 4;
      protocol.package_body_offset = 0;
      protocol.package_max_length = SW_INPUT_BUFFER_SIZE;
      protocol.onPackage = Client_onPackage;
  }
  
  int Client::sleep() {
      int ret;
      if (socket->events & SW_EVENT_WRITE) {
          ret = swoole_event_set(socket, SW_EVENT_WRITE);
      } else {
          ret = swoole_event_del(socket);
      }
      if (ret == SW_OK) {
          sleep_ = true;
      }
      return ret;
  }
  
  int Client::wakeup() {
      int ret;
      if (socket->events & SW_EVENT_WRITE) {
          ret = swoole_event_set(socket, SW_EVENT_READ | SW_EVENT_WRITE);
      } else {
          ret = swoole_event_add(socket, SW_EVENT_READ);
      }
      if (ret == SW_OK) {
          sleep_ = false;
      }
      return ret;
  }
  
  int Client::shutdown(int __how) {
      if (!socket || closed) {
          return SW_ERR;
      }
      if (__how == SHUT_RD) {
          if (shutdown_read || shutdow_rw || ::shutdown(socket->fd, SHUT_RD)) {
              return SW_ERR;
          } else {
              shutdown_read = 1;
              return SW_OK;
          }
      } else if (__how == SHUT_WR) {
          if (shutdown_write || shutdow_rw || ::shutdown(socket->fd, SHUT_WR) < 0) {
              return SW_ERR;
          } else {
              shutdown_write = 1;
              return SW_OK;
          }
      } else if (__how == SHUT_RDWR) {
          if (shutdow_rw || ::shutdown(socket->fd, SHUT_RDWR) < 0) {
              return SW_ERR;
          } else {
              shutdown_read = 1;
              return SW_OK;
          }
      } else {
          return SW_ERR;
      }
  }
  
  int Client::socks5_handshake(const char *recv_data, size_t length) {
      Socks5Proxy *ctx = socks5_proxy;
      char *buf = ctx->buf;
      uchar version, status, result, method;
  
      if (ctx->state == SW_SOCKS5_STATE_HANDSHAKE) {
          version = recv_data[0];
          method = recv_data[1];
          if (version != SW_SOCKS5_VERSION_CODE) {
              swoole_error_log(SW_LOG_NOTICE, SW_ERROR_SOCKS5_UNSUPPORT_VERSION, "SOCKS version is not supported");
              return SW_ERR;
          }
          if (method != ctx->method) {
              swoole_error_log(
                  SW_LOG_NOTICE, SW_ERROR_SOCKS5_UNSUPPORT_METHOD, "SOCKS authentication method is not supported");
              return SW_ERR;
          }
          // authenticate request
          if (method == SW_SOCKS5_METHOD_AUTH) {
              buf[0] = 0x01;
              buf[1] = ctx->username.length();
  
              buf += 2;
              memcpy(buf, ctx->username.c_str(), ctx->username.length());
              buf += ctx->username.length();
              buf[0] = ctx->password.length();
              memcpy(buf + 1, ctx->password.c_str(), ctx->password.length());
  
              ctx->state = SW_SOCKS5_STATE_AUTH;
  
              return send(this, ctx->buf, ctx->username.length() + ctx->password.length() + 3, 0);
          }
          // send connect request
          else {
          _send_connect_request:
              buf[0] = SW_SOCKS5_VERSION_CODE;
              buf[1] = 0x01;
              buf[2] = 0x00;
  
              ctx->state = SW_SOCKS5_STATE_CONNECT;
  
              if (ctx->dns_tunnel) {
                  buf[3] = 0x03;
                  buf[4] = ctx->target_host.length();
                  buf += 5;
                  memcpy(buf, ctx->target_host.c_str(), ctx->target_host.length());
                  buf += ctx->target_host.length();
                  *(uint16_t *) buf = htons(ctx->target_port);
                  return send(this, ctx->buf, ctx->target_host.length() + 7, 0);
              } else {
                  buf[3] = 0x01;
                  buf += 4;
                  *(uint32_t *) buf = htons(ctx->target_host.length());
                  buf += 4;
                  *(uint16_t *) buf = htons(ctx->target_port);
                  return send(this, ctx->buf, ctx->target_host.length() + 7, 0);
              }
          }
      } else if (ctx->state == SW_SOCKS5_STATE_AUTH) {
          version = recv_data[0];
          status = recv_data[1];
          if (version != 0x01) {
              swoole_error_log(SW_LOG_NOTICE, SW_ERROR_SOCKS5_UNSUPPORT_VERSION, "SOCKS version is not supported");
              return SW_ERR;
          }
          if (status != 0) {
              swoole_error_log(
                  SW_LOG_NOTICE, SW_ERROR_SOCKS5_AUTH_FAILED, "SOCKS username/password authentication failed");
              return SW_ERR;
          }
          goto _send_connect_request;
      } else if (ctx->state == SW_SOCKS5_STATE_CONNECT) {
          version = recv_data[0];
          if (version != SW_SOCKS5_VERSION_CODE) {
              swoole_error_log(SW_LOG_NOTICE, SW_ERROR_SOCKS5_UNSUPPORT_VERSION, "SOCKS version is not supported");
              return SW_ERR;
          }
          result = recv_data[1];
  #if 0
          uchar reg = recv_data[2];
          uchar type = recv_data[3];
          uint32_t ip = *(uint32_t *) (recv_data + 4);
          uint16_t port = *(uint16_t *) (recv_data + 8);
  #endif
          if (result == 0) {
              ctx->state = SW_SOCKS5_STATE_READY;
          } else {
              swoole_error_log(SW_LOG_NOTICE,
                               SW_ERROR_SOCKS5_SERVER_ERROR,
                               "Socks5 server error, reason :%s",
                               Socks5Proxy::strerror(result));
          }
          return result;
      }
      return SW_OK;
  }
  
  #ifdef SW_USE_OPENSSL
  int Client::enable_ssl_encrypt() {
      if (ssl_context) {
          return SW_ERR;
      }
      ssl_context.reset(new swoole::SSLContext());
      open_ssl = true;
      return SW_OK;
  }
  
  int Client::ssl_handshake() {
      if (socket->ssl) {
          return SW_ERR;
      }
      ssl_context->http_v2 = http2;
      if (!ssl_context->create()) {
          return SW_ERR;
      }
      socket->ssl_send_ = 1;
      if (socket->ssl_create(ssl_context.get(), SW_SSL_CLIENT) < 0) {
          return SW_ERR;
      }
  #ifdef SSL_CTRL_SET_TLSEXT_HOSTNAME
      if (!ssl_context->tls_host_name.empty()) {
          SSL_set_tlsext_host_name(socket->ssl, ssl_context->tls_host_name.c_str());
      }
  #endif
      if (socket->ssl_connect() < 0) {
          return SW_ERR;
      }
      if (socket->ssl_state == SW_SSL_STATE_READY && ssl_context->verify_peer) {
          if (ssl_verify(ssl_context->allow_self_signed) < 0) {
              return SW_ERR;
          }
      }
      return SW_OK;
  }
  
  int Client::ssl_verify(int allow_self_signed) {
      if (!socket->ssl_verify(allow_self_signed)) {
          return SW_ERR;
      }
  #ifdef SSL_CTRL_SET_TLSEXT_HOSTNAME
      if (!ssl_context->tls_host_name.empty() && !socket->ssl_check_host(ssl_context->tls_host_name.c_str())) {
          return SW_ERR;
      }
  #endif
      return SW_OK;
  }
  
  #endif
  
  static int Client_inet_addr(Client *cli, const char *host, int port) {
      // enable socks5 proxy
      if (cli->socks5_proxy) {
          cli->socks5_proxy->target_host = host;
          cli->socks5_proxy->target_port = port;
  
          host = cli->socks5_proxy->host.c_str();
          port = cli->socks5_proxy->port;
      }
  
      // enable http proxy
      if (cli->http_proxy) {
          cli->http_proxy->target_host = host;
          cli->http_proxy->target_port = port;
  
          host = cli->http_proxy->proxy_host.c_str();
          port = cli->http_proxy->proxy_port;
      }
  
      cli->server_host = host;
      cli->server_port = port;
  
      void *addr = nullptr;
      if (cli->socket->is_inet4()) {
          cli->server_addr.addr.inet_v4.sin_family = AF_INET;
          cli->server_addr.addr.inet_v4.sin_port = htons(port);
          cli->server_addr.len = sizeof(cli->server_addr.addr.inet_v4);
          addr = &cli->server_addr.addr.inet_v4.sin_addr.s_addr;
  
          if (inet_pton(AF_INET, host, addr)) {
              return SW_OK;
          }
      } else if (cli->socket->is_inet6()) {
          cli->server_addr.addr.inet_v6.sin6_family = AF_INET6;
          cli->server_addr.addr.inet_v6.sin6_port = htons(port);
          cli->server_addr.len = sizeof(cli->server_addr.addr.inet_v6);
          addr = cli->server_addr.addr.inet_v6.sin6_addr.s6_addr;
  
          if (inet_pton(AF_INET6, host, addr)) {
              return SW_OK;
          }
      } else if (cli->socket->is_local()) {
          cli->server_addr.addr.un.sun_family = AF_UNIX;
          swoole_strlcpy(cli->server_addr.addr.un.sun_path, host, sizeof(cli->server_addr.addr.un.sun_path));
          cli->server_addr.addr.un.sun_path[sizeof(cli->server_addr.addr.un.sun_path) - 1] = 0;
          cli->server_addr.len = sizeof(cli->server_addr.addr.un.sun_path);
          return SW_OK;
      } else {
          return SW_ERR;
      }
      if (!cli->async) {
          if (swoole::network::gethostbyname(cli->_sock_domain, host, (char *) addr) < 0) {
              swoole_set_last_error(SW_ERROR_DNSLOOKUP_RESOLVE_FAILED);
              return SW_ERR;
          }
      } else {
          cli->wait_dns = 1;
      }
      return SW_OK;
  }
  
  void Client::destroy() {
      if (destroyed) {
          return;
      }
      destroyed = true;
      swoole_event_defer(
          [](void *data) {
              Client *object = (Client *) data;
              delete object;
          },
          this);
  }
  
  Client::~Client() {
      if (!socket) {
          return;
      }
      assert(socket->fd != 0);
      // remove from reactor
      if (!closed) {
          close();
      }
      // clear buffer
      if (buffer) {
          delete buffer;
          buffer = nullptr;
      }
      if (server_str) {
          ::sw_free((void *) server_str);
      }
      if (socket->out_buffer) {
          delete socket->out_buffer;
          socket->out_buffer = nullptr;
      }
      if (socket->in_buffer) {
          delete socket->in_buffer;
          socket->in_buffer = nullptr;
      }
      if (socks5_proxy) {
          delete socks5_proxy;
      }
      if (http_proxy) {
          delete http_proxy;
      }
      if (async) {
          socket->free();
      } else {
          delete socket;
      }
  }
  
  int Client::close() {
      if (socket == nullptr || closed) {
          return SW_ERR;
      }
      closed = 1;
  
      int fd = socket->fd;
      assert(fd != 0);
  
  #ifdef SW_USE_OPENSSL
      if (open_ssl && ssl_context) {
          if (socket->ssl) {
              socket->ssl_close();
          }
      }
  #endif
  
      if (socket->socket_type == SW_SOCK_UNIX_DGRAM) {
          unlink(socket->info.addr.un.sun_path);
      }
      if (async) {
          // remove from reactor
          if (!socket->removed) {
              swoole_event_del(socket);
          }
          if (timer) {
              swoole_timer_del(timer);
              timer = nullptr;
          }
          // onClose callback
          if (active && onClose) {
              active = 0;
              onClose(this);
          }
      } else {
          active = 0;
      }
  
      /**
       * fd marked -1, prevent double close
       */
      socket->fd = -1;
  
      return ::close(fd);
  }
  
  static int Client_tcp_connect_sync(Client *cli, const char *host, int port, double timeout, int nonblock) {
      int ret;
  
      cli->timeout = timeout;
  
      if (Client_inet_addr(cli, host, port) < 0) {
          return SW_ERR;
      }
  
      if (nonblock) {
          cli->socket->set_nonblock();
      } else {
          if (cli->timeout > 0) {
              cli->socket->set_timeout(timeout);
          }
  #ifndef HAVE_KQUEUE
          cli->socket->set_block();
  #endif
      }
      while (1) {
  #ifdef HAVE_KQUEUE
          if (nonblock == 2) {
              // special case on MacOS
              ret = cli->socket->connect(cli->server_addr);
          } else {
              cli->socket->set_nonblock();
              ret = cli->socket->connect(cli->server_addr);
              if (ret < 0) {
                  if (errno != EINPROGRESS) {
                      return SW_ERR;
                  }
                  if (cli->socket->wait_event(timeout > 0 ? (int) (timeout * 1000) : timeout, SW_EVENT_WRITE) < 0) {
                      swoole_set_last_error(ETIMEDOUT);
                      return SW_ERR;
                  }
                  int err;
                  socklen_t len = sizeof(len);
                  ret = cli->socket->get_option(SOL_SOCKET, SO_ERROR, &err, &len);
                  if (ret < 0) {
                      swoole_set_last_error(errno);
                      return SW_ERR;
                  }
                  if (err != 0) {
                      swoole_set_last_error(err);
                      return SW_ERR;
                  }
                  cli->socket->set_block();
                  ret = 0;
              }
          }
  #else
          ret = cli->socket->connect(cli->server_addr);
          if (ret < 0 && errno == EINPROGRESS) {
              errno = ETIMEDOUT;
          }
  #endif
          if (ret < 0) {
              if (errno == EINTR) {
                  continue;
              }
              swoole_set_last_error(errno);
          }
          break;
      }
  
      if (ret >= 0) {
          cli->active = 1;
  
          // socks5 proxy
          if (cli->socks5_proxy) {
              char buf[1024];
              Socks5Proxy::pack(buf, cli->socks5_proxy->username.empty() ? 0x00 : 0x02);
              if (cli->send(cli, buf, 3, 0) < 0) {
                  return SW_ERR;
              }
              cli->socks5_proxy->state = SW_SOCKS5_STATE_HANDSHAKE;
              while (1) {
                  ssize_t n = cli->recv(cli, buf, sizeof(buf), 0);
                  if (n > 0) {
                      if (cli->socks5_handshake(buf, n) < 0) {
                          return SW_ERR;
                      } else {
                          if (cli->socks5_proxy->state == SW_SOCKS5_STATE_READY) {
                              break;
                          } else {
                              continue;
                          }
                      }
                  }
                  return SW_ERR;
              }
          }
  
  #ifdef SW_USE_OPENSSL
          if (cli->open_ssl) {
              if (cli->ssl_handshake() < 0) {
                  return SW_ERR;
              }
          }
  #endif
      }
  
      return ret;
  }
  
  static int Client_tcp_connect_async(Client *cli, const char *host, int port, double timeout, int nonblock) {
      int ret;
  
      cli->timeout = timeout;
  
      if (!cli->buffer) {
          cli->buffer = new String(cli->input_buffer_size);
      }
  
      if (!(cli->onConnect && cli->onError && cli->onClose)) {
          swWarn("onConnect/onError/onClose callback have not set");
          return SW_ERR;
      }
  
      if (cli->onBufferFull && cli->buffer_high_watermark == 0) {
          cli->buffer_high_watermark = cli->socket->buffer_size * 0.8;
      }
  
      if (Client_inet_addr(cli, host, port) < 0) {
          return SW_ERR;
      }
  
      if (cli->wait_dns) {
          AsyncEvent ev{};
  
          size_t len = strlen(cli->server_host);
          if (len < SW_IP_MAX_LENGTH) {
              ev.nbytes = SW_IP_MAX_LENGTH;
          } else {
              ev.nbytes = len + 1;
          }
  
          ev.buf = sw_malloc(ev.nbytes);
          if (!ev.buf) {
              swWarn("malloc failed");
              return SW_ERR;
          }
  
          memcpy(ev.buf, cli->server_host, len);
          ((char *) ev.buf)[len] = 0;
          ev.flags = cli->_sock_domain;
          ev.object = cli;
          ev.fd = cli->socket->fd;
          ev.handler = async::handler_gethostbyname;
          ev.callback = Client_onResolveCompleted;
  
          if (swoole::async::dispatch(&ev) == nullptr) {
              sw_free(ev.buf);
              return SW_ERR;
          } else {
              return SW_OK;
          }
      }
  
      while (1) {
          ret = cli->socket->connect(cli->server_addr);
          if (ret < 0) {
              if (errno == EINTR) {
                  continue;
              }
              swoole_set_last_error(errno);
          }
          break;
      }
  
      if ((ret < 0 && errno == EINPROGRESS) || ret == 0) {
          if (swoole_event_add(cli->socket, SW_EVENT_WRITE) < 0) {
              return SW_ERR;
          }
          if (timeout > 0) {
              cli->timer = swoole_timer_add((long) (timeout * 1000), false, Client_onTimeout, cli);
          }
          return SW_OK;
      } else {
          cli->active = 0;
          cli->socket->removed = 1;
          cli->close();
          if (cli->onError) {
              cli->onError(cli);
          }
      }
  
      return ret;
  }
  
  static ssize_t Client_tcp_send_async(Client *cli, const char *data, size_t length, int flags) {
      int n = length;
      if (swoole_event_write(cli->socket, data, length) < 0) {
          if (swoole_get_last_error() == SW_ERROR_OUTPUT_BUFFER_OVERFLOW) {
              n = -1;
              cli->high_watermark = 1;
          } else {
              return SW_ERR;
          }
      }
      if (cli->onBufferFull && cli->socket->out_buffer && cli->high_watermark == 0 &&
          cli->socket->out_buffer->length() >= cli->buffer_high_watermark) {
          cli->high_watermark = 1;
          cli->onBufferFull(cli);
      }
      return n;
  }
  
  static ssize_t Client_tcp_send_sync(Client *cli, const char *data, size_t length, int flags) {
      size_t written = 0;
      ssize_t n;
  
      assert(length > 0);
      assert(data != nullptr);
  
      while (written < length) {
          n = cli->socket->send(data, length - written, flags);
          if (n < 0) {
              if (errno == EINTR) {
                  continue;
              } else if (errno == EAGAIN) {
                  cli->socket->wait_event(1000, SW_EVENT_WRITE);
                  continue;
              } else {
                  swoole_set_last_error(errno);
                  return SW_ERR;
              }
          }
          written += n;
          data += n;
      }
  
      return written;
  }
  
  static int Client_tcp_sendfile_sync(Client *cli, const char *filename, off_t offset, size_t length) {
      if (cli->socket->sendfile_blocking(filename, offset, length, cli->timeout) < 0) {
          swoole_set_last_error(errno);
          return SW_ERR;
      }
      return SW_OK;
  }
  
  static int Client_tcp_sendfile_async(Client *cli, const char *filename, off_t offset, size_t length) {
      if (cli->socket->sendfile(filename, offset, length) < 0) {
          swoole_set_last_error(errno);
          return SW_ERR;
      }
      if (swoole_event_add_or_update(cli->socket, SW_EVENT_WRITE) == SW_ERR) {
          return SW_ERR;
      }
      return SW_OK;
  }
  
  /**
   * Only for synchronous client
   */
  static ssize_t Client_tcp_recv_no_buffer(Client *cli, char *data, size_t len, int flag) {
      ssize_t ret;
  
      while (1) {
  #ifdef HAVE_KQUEUE
          int timeout_ms = (int) (cli->timeout * 1000);
          if (cli->socket->wait_event(timeout_ms, SW_EVENT_READ) < 0) {
              return -1;
          }
  #endif
          ret = cli->socket->recv(data, len, flag);
          if (ret >= 0) {
              break;
          }
          if (errno == EINTR) {
              if (cli->interrupt_time <= 0) {
                  cli->interrupt_time = microtime();
                  continue;
              } else if (microtime() > cli->interrupt_time + cli->timeout) {
                  break;
              } else {
                  continue;
              }
          }
  #ifdef SW_USE_OPENSSL
          if (errno == EAGAIN && cli->socket->ssl) {
              int timeout_ms = (int) (cli->timeout * 1000);
              if (cli->socket->ssl_want_read && cli->socket->wait_event(timeout_ms, SW_EVENT_READ) == SW_OK) {
                  continue;
              } else if (cli->socket->ssl_want_write && cli->socket->wait_event(timeout_ms, SW_EVENT_WRITE) == SW_OK) {
                  continue;
              }
          }
  #endif
          break;
      }
  
      return ret;
  }
  
  static int Client_udp_connect(Client *cli, const char *host, int port, double timeout, int udp_connect) {
      if (Client_inet_addr(cli, host, port) < 0) {
          return SW_ERR;
      }
  
      cli->active = 1;
      cli->timeout = timeout;
      int bufsize = Socket::default_buffer_size;
  
      if (timeout > 0) {
          cli->socket->set_timeout(timeout);
      }
  
      if (cli->socket->socket_type == SW_SOCK_UNIX_DGRAM) {
          struct sockaddr_un *client_addr = &cli->socket->info.addr.un;
          sprintf(client_addr->sun_path, "/tmp/swoole-client.%d.%d.sock", getpid(), cli->socket->fd);
          client_addr->sun_family = AF_UNIX;
          unlink(client_addr->sun_path);
  
          if (bind(cli->socket->fd, (struct sockaddr *) client_addr, sizeof(cli->socket->info.addr.un)) < 0) {
              swSysWarn("bind(%s) failed", client_addr->sun_path);
              return SW_ERR;
          }
      }
  
  #ifdef SW_USE_OPENSSL
      if (cli->open_ssl)
  #ifdef SW_SUPPORT_DTLS
      {
          udp_connect = 1;
          cli->ssl_context->protocols = SW_SSL_DTLS;
          cli->socket->dtls = 1;
          cli->socket->chunk_size = SW_SSL_BUFFER_SIZE;
          cli->send = Client_tcp_send_sync;
          cli->recv = Client_tcp_recv_no_buffer;
      }
  #else
      {
          swWarn("DTLS support require openssl-1.1 or later");
          return SW_ERR;
      }
  #endif
  #endif
  
      if (udp_connect != 1) {
          goto _connect_ok;
      }
  
      if (cli->socket->connect(cli->server_addr) == 0) {
          cli->socket->clean();
      _connect_ok:
  
          cli->socket->set_option(SOL_SOCKET, SO_SNDBUF, bufsize);
          cli->socket->set_option(SOL_SOCKET, SO_RCVBUF, bufsize);
  
          if (cli->async && cli->onConnect) {
              if (swoole_event_add(cli->socket, SW_EVENT_READ) < 0) {
                  return SW_ERR;
              }
              execute_onConnect(cli);
          }
  #ifdef SW_USE_OPENSSL
          if (cli->open_ssl && cli->ssl_handshake() < 0) {
              return SW_ERR;
          }
  #endif
          return SW_OK;
      } else {
          cli->active = 0;
          cli->socket->removed = 1;
          cli->close();
          if (cli->async && cli->onError) {
              cli->onError(cli);
          }
          return SW_ERR;
      }
  }
  
  static ssize_t Client_udp_send(Client *cli, const char *data, size_t len, int flags) {
      ssize_t n = sendto(cli->socket->fd, data, len, 0, (struct sockaddr *) &cli->server_addr.addr, cli->server_addr.len);
      if (n < 0 || n < (ssize_t) len) {
          return SW_ERR;
      } else {
          return n;
      }
  }
  
  static ssize_t Client_udp_recv(Client *cli, char *data, size_t length, int flags) {
  #ifdef HAVE_KQUEUE
      if (!cli->async) {
          int timeout_ms = (int) (cli->timeout * 1000);
          if (cli->socket->wait_event(timeout_ms, SW_EVENT_READ) < 0) {
              return -1;
          }
      }
  #endif
      ssize_t ret = cli->socket->recvfrom(data, length, flags, &cli->remote_addr);
      if (ret < 0) {
          if (errno == EINTR) {
              ret = cli->socket->recvfrom(data, length, flags, &cli->remote_addr);
          } else {
              return SW_ERR;
          }
      }
      return ret;
  }
  
  #ifdef SW_USE_OPENSSL
  static int Client_https_proxy_handshake(Client *cli) {
      char *buf = cli->buffer->str;
      size_t len = cli->buffer->length;
      int state = 0;
      char *p = buf;
      char *pe = buf + len;
      for (; p < pe; p++) {
          if (state == 0) {
              if (SW_STRCASECT(p, pe - p, "HTTP/1.1") || SW_STRCASECT(p, pe - p, "HTTP/1.0")) {
                  state = 1;
                  p += sizeof("HTTP/1.x") - 1;
              } else {
                  break;
              }
          } else if (state == 1) {
              if (isspace(*p)) {
                  continue;
              } else {
                  if (SW_STRCASECT(p, pe - p, "200")) {
                      state = 2;
                      p += sizeof("200") - 1;
                  } else {
                      break;
                  }
              }
          } else if (state == 2) {
              if (isspace(*p)) {
                  continue;
              } else {
                  if (SW_STRCASECT(p, pe - p, "Connection established")) {
                      return SW_OK;
                  } else {
                      break;
                  }
              }
          }
      }
      return SW_ERR;
  }
  #endif
  
  static int Client_onPackage(Protocol *proto, Socket *conn, const char *data, uint32_t length) {
      Client *cli = (Client *) conn->object;
      cli->onReceive(cli, data, length);
      return conn->close_wait ? SW_ERR : SW_OK;
  }
  
  static int Client_onStreamRead(Reactor *reactor, Event *event) {
      ssize_t n = -1;
      Client *cli = (Client *) event->socket->object;
      char *buf = cli->buffer->str + cli->buffer->length;
      ssize_t buf_size = cli->buffer->size - cli->buffer->length;
  
      if (cli->http_proxy && cli->http_proxy->state != SW_HTTP_PROXY_STATE_READY) {
  #ifdef SW_USE_OPENSSL
          if (cli->open_ssl) {
              n = event->socket->recv(buf, buf_size, 0);
              if (n <= 0) {
                  goto __close;
              }
              cli->buffer->length += n;
              if (cli->buffer->length < sizeof(SW_HTTPS_PROXY_HANDSHAKE_RESPONSE) - 1) {
                  return SW_OK;
              }
              if (Client_https_proxy_handshake(cli) < 0) {
                  swoole_error_log(
                      SW_LOG_NOTICE, SW_ERROR_HTTP_PROXY_HANDSHAKE_ERROR, "failed to handshake with http proxy");
                  goto _connect_fail;
              } else {
                  cli->http_proxy->state = SW_HTTP_PROXY_STATE_READY;
                  cli->buffer->clear();
              }
              if (cli->ssl_handshake() < 0) {
                  goto _connect_fail;
              } else {
                  cli->socket->ssl_state = SW_SSL_STATE_WAIT_STREAM;
                  return swoole_event_set(event->socket, SW_EVENT_WRITE);
              }
              if (cli->onConnect) {
                  execute_onConnect(cli);
              }
              return SW_OK;
          }
  #endif
      }
      if (cli->socks5_proxy && cli->socks5_proxy->state != SW_SOCKS5_STATE_READY) {
          n = event->socket->recv(buf, buf_size, 0);
          if (n <= 0) {
              goto __close;
          }
          if (cli->socks5_handshake(buf, buf_size) < 0) {
              goto __close;
          }
          if (cli->socks5_proxy->state != SW_SOCKS5_STATE_READY) {
              return SW_OK;
          }
  #ifdef SW_USE_OPENSSL
          if (cli->open_ssl) {
              if (cli->ssl_handshake() < 0) {
                  _connect_fail: cli->active = 0;
                  cli->close();
                  if (cli->onError) {
                      cli->onError(cli);
                  }
              } else {
                  cli->socket->ssl_state = SW_SSL_STATE_WAIT_STREAM;
                  return swoole_event_set(event->socket, SW_EVENT_WRITE);
              }
          } else
  #endif
          {
              if (cli->onConnect) {
                  execute_onConnect(cli);
              }
          }
          return SW_OK;
      }
  
  #ifdef SW_USE_OPENSSL
      if (cli->open_ssl && cli->socket->ssl_state == SW_SSL_STATE_WAIT_STREAM) {
          if (cli->ssl_handshake() < 0) {
              goto _connect_fail;
          }
          if (cli->socket->ssl_state != SW_SSL_STATE_READY) {
              return SW_OK;
          }
          // ssl handshake sucess
          else if (cli->onConnect) {
              execute_onConnect(cli);
          }
      }
  #endif
  
      if (cli->open_eof_check || cli->open_length_check) {
          Socket *conn = cli->socket;
          Protocol *protocol = &cli->protocol;
  
          if (cli->open_eof_check) {
              n = protocol->recv_with_eof_protocol(conn, cli->buffer);
          } else {
              n = protocol->recv_with_length_protocol(conn, cli->buffer);
          }
  
          if (n < 0) {
              if (!cli->closed) {
                  cli->close();
              }
              return SW_OK;
          } else {
              if (conn->removed == 0 && cli->remove_delay) {
                  cli->sleep();
                  cli->remove_delay = 0;
              }
              return SW_OK;
          }
      }
  
  #ifdef SW_CLIENT_RECV_AGAIN
  _recv_again:
  #endif
      n = event->socket->recv(buf, buf_size, 0);
      if (n < 0) {
          switch (event->socket->catch_error(errno)) {
          case SW_ERROR:
              swSysWarn("Read from socket[%d] failed", event->fd);
              return SW_OK;
          case SW_CLOSE:
              goto __close;
          case SW_WAIT:
              return SW_OK;
          default:
              return SW_OK;
          }
      } else if (n == 0) {
      __close:
          return cli->close();
      } else {
          cli->onReceive(cli, buf, n);
  #ifdef SW_CLIENT_RECV_AGAIN
          if (n == buf_size) {
              goto _recv_again;
          }
  #endif
          return SW_OK;
      }
      return SW_OK;
  }
  
  static int Client_onDgramRead(Reactor *reactor, Event *event) {
      Client *cli = (Client *) event->socket->object;
      char buffer[SW_BUFFER_SIZE_UDP];
  
      int n = Client_udp_recv(cli, buffer, sizeof(buffer), 0);
      if (n < 0) {
          return SW_ERR;
      } else {
          cli->onReceive(cli, buffer, n);
      }
      return SW_OK;
  }
  
  static int Client_onError(Reactor *reactor, Event *event) {
      Client *cli = (Client *) event->socket->object;
      if (cli->active) {
          return cli->close();
      } else {
          Client_onWrite(reactor, event);
      }
      return SW_OK;
  }
  
  static void Client_onTimeout(Timer *timer, TimerNode *tnode) {
      Client *cli = (Client *) tnode->data;
      swoole_set_last_error(ETIMEDOUT);
  
  #ifdef SW_USE_OPENSSL
      if (cli->open_ssl && cli->socket->ssl_state != SW_SSL_STATE_READY) {
          cli->active = 0;
      }
  #endif
      if (cli->socks5_proxy && cli->socks5_proxy->state != SW_SOCKS5_STATE_READY) {
          cli->active = 0;
      } else if (cli->http_proxy && cli->http_proxy->state != SW_HTTP_PROXY_STATE_READY) {
          cli->active = 0;
      }
  
      cli->close();
      if (cli->onError) {
          cli->onError(cli);
      }
  }
  
  static void Client_onResolveCompleted(AsyncEvent *event) {
      if (event->canceled) {
          sw_free(event->buf);
          return;
      }
  
      Client *cli = (Client *) event->object;
      cli->wait_dns = 0;
  
      if (event->error == 0) {
          Client_tcp_connect_async(cli, (char *) event->buf, cli->server_port, cli->timeout, 1);
      } else {
          swoole_set_last_error(SW_ERROR_DNSLOOKUP_RESOLVE_FAILED);
          cli->socket->removed = 1;
          cli->close();
          if (cli->onError) {
              cli->onError(cli);
          }
      }
      sw_free(event->buf);
  }
  
  static int Client_onWrite(Reactor *reactor, Event *event) {
      Client *cli = (Client *) event->socket->object;
      Socket *_socket = cli->socket;
      int ret;
      int err;
  
      if (cli->active) {
  #ifdef SW_USE_OPENSSL
          if (cli->open_ssl && _socket->ssl_state == SW_SSL_STATE_WAIT_STREAM) {
              if (cli->ssl_handshake() < 0) {
                  goto _connect_fail;
              } else if (_socket->ssl_state == SW_SSL_STATE_READY) {
                  goto _connect_success;
              } else {
                  if (_socket->ssl_want_read) {
                      swoole_event_set(event->socket, SW_EVENT_READ);
                  }
                  return SW_OK;
              }
          }
  #endif
          if (Reactor::_writable_callback(reactor, event) < 0) {
              return SW_ERR;
          }
          if (cli->onBufferEmpty && cli->high_watermark && _socket->out_buffer->length() <= cli->buffer_low_watermark) {
              cli->high_watermark = 0;
              cli->onBufferEmpty(cli);
          }
          return SW_OK;
      }
  
      ret = _socket->get_option(SOL_SOCKET, SO_ERROR, &err);
      swoole_set_last_error(err);
      if (ret < 0) {
          swSysWarn("getsockopt(%d) failed", event->fd);
          return SW_ERR;
      }
  
      // success
      if (swoole_get_last_error() == 0) {
          // listen read event
          swoole_event_set(event->socket, SW_EVENT_READ);
          // connected
          cli->active = 1;
          // socks5 proxy
          if (cli->socks5_proxy && cli->socks5_proxy->state == SW_SOCKS5_STATE_WAIT) {
              char buf[3];
              Socks5Proxy::pack(buf, cli->socks5_proxy->username.empty() ? 0x00 : 0x02);
              cli->socks5_proxy->state = SW_SOCKS5_STATE_HANDSHAKE;
              return cli->send(cli, buf, sizeof(buf), 0);
          }
          // http proxy
          if (cli->http_proxy && cli->http_proxy->state == SW_HTTP_PROXY_STATE_WAIT) {
  #ifdef SW_USE_OPENSSL
              if (cli->open_ssl) {
                  cli->http_proxy->state = SW_HTTP_PROXY_STATE_HANDSHAKE;
                  int n = sw_snprintf(cli->http_proxy->buf,
                                      sizeof(cli->http_proxy->buf),
                                      "CONNECT %s:%d HTTP/1.1\r\n\r\n",
                                      cli->http_proxy->target_host.c_str(),
                                      cli->http_proxy->target_port);
                  return cli->send(cli, cli->http_proxy->buf, n, 0);
              }
  #endif
          }
  #ifdef SW_USE_OPENSSL
          if (cli->open_ssl) {
              if (cli->ssl_handshake() < 0) {
                  goto _connect_fail;
              } else {
                  _socket->ssl_state = SW_SSL_STATE_WAIT_STREAM;
              }
              return SW_OK;
          }
      _connect_success:
  #endif
          if (cli->onConnect) {
              execute_onConnect(cli);
          }
      } else {
  #ifdef SW_USE_OPENSSL
      _connect_fail:
  #endif
          cli->active = 0;
          cli->close();
          if (cli->onError) {
              cli->onError(cli);
          }
      }
  
      return SW_OK;
  }
  
  }  // namespace network
  }  // namespace swoole
  