static VALUE _mysql_client_options(VALUE self, int opt, VALUE value) {
  int result;
  void *retval = NULL;
  unsigned int intval = 0;
  my_bool boolval;

  GET_CLIENT(self);

  REQUIRE_NOT_CONNECTED(wrapper);

  if (NIL_P(value))
      return Qfalse;

  switch(opt) {
    case MYSQL_OPT_CONNECT_TIMEOUT:
      intval = NUM2INT(value);
      retval = &intval;
      break;

    case MYSQL_OPT_READ_TIMEOUT:
      intval = NUM2INT(value);
      retval = &intval;
      break;

    case MYSQL_OPT_WRITE_TIMEOUT:
      intval = NUM2INT(value);
      retval = &intval;
      break;

    case MYSQL_OPT_LOCAL_INFILE:
      intval = (value == Qfalse ? 0 : 1);
      retval = &intval;
      break;

    case MYSQL_OPT_RECONNECT:
      boolval = (value == Qfalse ? 0 : 1);
      retval = &boolval;
      break;

    default:
      return Qfalse;
  }

  result = mysql_options(wrapper->client, opt, retval);

  /* Zero means success */
  if (result != 0) {
    rb_warn("%s\n", mysql_error(wrapper->client));
  } else {
    /* Special case for reconnect, this option is also stored in the wrapper struct */
    if (opt == MYSQL_OPT_RECONNECT)
      wrapper->reconnect_enabled = boolval;
  }

  return (result == 0) ? Qtrue : Qfalse;
}

VALUE rb_mysql_stmt_new(VALUE rb_client, VALUE sql) {
  mysql_stmt_wrapper *stmt_wrapper;
  VALUE rb_stmt;
#ifdef HAVE_RUBY_ENCODING_H
  rb_encoding *conn_enc;
#endif

  Check_Type(sql, T_STRING);

  rb_stmt = Data_Make_Struct(cMysql2Statement, mysql_stmt_wrapper, rb_mysql_stmt_mark, rb_mysql_stmt_free, stmt_wrapper);
  {
    stmt_wrapper->client = rb_client;
    stmt_wrapper->refcount = 1;
    stmt_wrapper->closed = 0;
    stmt_wrapper->stmt = NULL;
  }

  // instantiate stmt
  {
    GET_CLIENT(rb_client);
    stmt_wrapper->stmt = mysql_stmt_init(wrapper->client);
#ifdef HAVE_RUBY_ENCODING_H
    conn_enc = rb_to_encoding(wrapper->encoding);
#endif
  }
  if (stmt_wrapper->stmt == NULL) {
    rb_raise(cMysql2Error, "Unable to initialize prepared statement: out of memory");
  }

  // set STMT_ATTR_UPDATE_MAX_LENGTH attr
  {
    my_bool truth = 1;
    if (mysql_stmt_attr_set(stmt_wrapper->stmt, STMT_ATTR_UPDATE_MAX_LENGTH, &truth)) {
      rb_raise(cMysql2Error, "Unable to initialize prepared statement: set STMT_ATTR_UPDATE_MAX_LENGTH");
    }
  }

  // call mysql_stmt_prepare w/o gvl
  {
    struct nogvl_prepare_statement_args args;
    args.stmt = stmt_wrapper->stmt;
    args.sql = sql;
#ifdef HAVE_RUBY_ENCODING_H
    // ensure the string is in the encoding the connection is expecting
    args.sql = rb_str_export_to_enc(args.sql, conn_enc);
#endif
    args.sql_ptr = RSTRING_PTR(sql);
    args.sql_len = RSTRING_LEN(sql);

    if ((VALUE)rb_thread_call_without_gvl(nogvl_prepare_statement, &args, RUBY_UBF_IO, 0) == Qfalse) {
      rb_raise_mysql2_stmt_error(stmt_wrapper);
    }
  }

  return rb_stmt;
}

static VALUE rb_mysql_client_next_result(VALUE self)
{
    GET_CLIENT(self);
    int ret;
    ret = mysql_next_result(wrapper->client);
    if (ret == 0)
      return Qtrue;
    else
      return Qfalse;
}

static VALUE rb_mysql_client_socket(VALUE self) {
  GET_CLIENT(self);
#ifndef _WIN32
  REQUIRE_OPEN_DB(wrapper);
  int fd_set_fd = wrapper->client->net.fd;
  return INT2NUM(fd_set_fd);
#else
  rb_raise(cMysql2Error, "Raw access to the mysql file descriptor isn't supported on Windows");
#endif
}

static VALUE initialize_ext(VALUE self) {
  GET_CLIENT(self);

  if ((VALUE)rb_thread_call_without_gvl(nogvl_init, wrapper, RUBY_UBF_IO, 0) == Qfalse) {
    /* TODO: warning - not enough memory? */
    return rb_raise_mysql2_error(wrapper);
  }

  wrapper->initialized = 1;
  return self;
}

static VALUE init_connection(VALUE self) {
  GET_CLIENT(self);

  if (rb_thread_blocking_region(nogvl_init, wrapper->client, RUBY_UBF_IO, 0) == Qfalse) {
    /* TODO: warning - not enough memory? */
    return rb_raise_mysql2_error(wrapper);
  }

  wrapper->closed = 0;
  return self;
}

static VALUE rb_mysql_client_affected_rows(VALUE self) {
  my_ulonglong retVal;
  GET_CLIENT(self);

  REQUIRE_OPEN_DB(wrapper);
  retVal = mysql_affected_rows(wrapper->client);
  if (retVal == (my_ulonglong)-1) {
    rb_raise_mysql2_error(wrapper);
  }
  return ULL2NUM(retVal);
}

static VALUE set_ssl_options(VALUE self, VALUE key, VALUE cert, VALUE ca, VALUE capath, VALUE cipher) {
  GET_CLIENT(self);

  mysql_ssl_set(wrapper->client,
      NIL_P(key)    ? NULL : StringValueCStr(key),
      NIL_P(cert)   ? NULL : StringValueCStr(cert),
      NIL_P(ca)     ? NULL : StringValueCStr(ca),
      NIL_P(capath) ? NULL : StringValueCStr(capath),
      NIL_P(cipher) ? NULL : StringValueCStr(cipher));

  return self;
}

/* call-seq:
 *    client.automatic_close = false
 *
 * Set this to +false+ to leave the connection open after it is garbage
 * collected. To avoid "Aborted connection" errors on the server, explicitly
 * call +close+ when the connection is no longer needed.
 *
 * @see http://dev.mysql.com/doc/en/communication-errors.html
 */
static VALUE set_automatic_close(VALUE self, VALUE value) {
  GET_CLIENT(self);
  if (RTEST(value)) {
    wrapper->automatic_close = 1;
  } else {
#ifndef _WIN32
    wrapper->automatic_close = 0;
#else
    rb_warn("Connections are always closed by garbage collector on Windows");
#endif
  }
  return value;
}

/* call-seq:
 *    client.next_result
 *
 * Fetch the next result set from the server.
 * Returns nothing.
 */
static VALUE rb_mysql_client_next_result(VALUE self)
{
    int ret;
    GET_CLIENT(self);
    ret = mysql_next_result(wrapper->client);
    if (ret > 0) {
      rb_raise_mysql2_error(wrapper);
      return Qfalse;
    } else if (ret == 0) {
      return Qtrue;
    } else {
      return Qfalse;
    }
}

static VALUE disconnect_and_raise(VALUE self, VALUE error) {
  GET_CLIENT(self);

  wrapper->closed = 1;
  wrapper->active = 0;

  // manually close the socket for read/write
  // this feels dirty, but is there another way?
  shutdown(wrapper->client->net.fd, 2);

  rb_exc_raise(error);

  return Qnil;
}

static VALUE rb_connect(VALUE self, VALUE user, VALUE pass, VALUE host, VALUE port, VALUE database, VALUE socket, VALUE flags) {
  struct nogvl_connect_args args;
  time_t start_time, end_time;
  unsigned int elapsed_time, connect_timeout;
  VALUE rv;
  GET_CLIENT(self);

  args.host        = NIL_P(host)     ? NULL : StringValueCStr(host);
  args.unix_socket = NIL_P(socket)   ? NULL : StringValueCStr(socket);
  args.port        = NIL_P(port)     ? 0    : NUM2INT(port);
  args.user        = NIL_P(user)     ? NULL : StringValueCStr(user);
  args.passwd      = NIL_P(pass)     ? NULL : StringValueCStr(pass);
  args.db          = NIL_P(database) ? NULL : StringValueCStr(database);
  args.mysql       = wrapper->client;
  args.client_flag = NUM2ULONG(flags);

  if (wrapper->connect_timeout)
    time(&start_time);
  rv = (VALUE) rb_thread_call_without_gvl(nogvl_connect, &args, RUBY_UBF_IO, 0);
  if (rv == Qfalse) {
    while (rv == Qfalse && errno == EINTR) {
      if (wrapper->connect_timeout) {
        time(&end_time);
        /* avoid long connect timeout from system time changes */
        if (end_time < start_time)
          start_time = end_time;
        elapsed_time = end_time - start_time;
        /* avoid an early timeout due to time truncating milliseconds off the start time */
        if (elapsed_time > 0)
          elapsed_time--;
        if (elapsed_time >= wrapper->connect_timeout)
          break;
        connect_timeout = wrapper->connect_timeout - elapsed_time;
        mysql_options(wrapper->client, MYSQL_OPT_CONNECT_TIMEOUT, &connect_timeout);
      }
      errno = 0;
      rv = (VALUE) rb_thread_call_without_gvl(nogvl_connect, &args, RUBY_UBF_IO, 0);
    }
    /* restore the connect timeout for reconnecting */
    if (wrapper->connect_timeout)
      mysql_options(wrapper->client, MYSQL_OPT_CONNECT_TIMEOUT, &wrapper->connect_timeout);
    if (rv == Qfalse)
      return rb_raise_mysql2_error(wrapper);
  }

  wrapper->server_version = mysql_get_server_version(wrapper->client);
  wrapper->connected = 1;
  return self;
}

static VALUE disconnect_and_raise(VALUE self, VALUE error) {
  GET_CLIENT(self);

  wrapper->active_thread = Qnil;
  wrapper->connected = 0;

  /* manually close the socket for read/write
     this feels dirty, but is there another way? */
  close(wrapper->client->net.fd);
  wrapper->client->net.fd = -1;

  rb_exc_raise(error);

  return Qnil;
}

static VALUE set_reconnect(VALUE self, VALUE value) {
    my_bool reconnect;
    GET_CLIENT(self);

    if(!NIL_P(value)) {
        reconnect = value == Qfalse ? 0 : 1;

        /* set default reconnect behavior */
        if (mysql_options(wrapper->client, MYSQL_OPT_RECONNECT, &reconnect)) {
            /* TODO: warning - unable to set reconnect behavior */
            rb_warn("%s\n", mysql_error(wrapper->client));
        }
    }
    return value;
}

static VALUE rb_set_ssl_mode_option(VALUE self, VALUE setting) {
  unsigned long version = mysql_get_client_version();

  if (version < 50703) {
    rb_warn( "Your mysql client library does not support setting ssl_mode; full support comes with 5.7.11." );
    return Qnil;
  }
#ifdef HAVE_CONST_MYSQL_OPT_SSL_ENFORCE
  GET_CLIENT(self);
  int val = NUM2INT( setting );
  if (version >= 50703 && version < 50711) {
    if (val == SSL_MODE_DISABLED || val == SSL_MODE_REQUIRED) {
      bool b = ( val == SSL_MODE_REQUIRED );
      int result = mysql_options( wrapper->client, MYSQL_OPT_SSL_ENFORCE, &b );
      return INT2NUM(result);
    } else {
      rb_warn( "MySQL client libraries between 5.7.3 and 5.7.10 only support SSL_MODE_DISABLED and SSL_MODE_REQUIRED" );
      return Qnil;
    }
  }
#endif
#ifdef FULL_SSL_MODE_SUPPORT
  GET_CLIENT(self);
  int val = NUM2INT( setting );

  if (val != SSL_MODE_DISABLED && val != SSL_MODE_PREFERRED && val != SSL_MODE_REQUIRED && val != SSL_MODE_VERIFY_CA && val != SSL_MODE_VERIFY_IDENTITY) {
    rb_raise(cMysql2Error, "ssl_mode= takes DISABLED, PREFERRED, REQUIRED, VERIFY_CA, VERIFY_IDENTITY, you passed: %d", val );
  }
  int result = mysql_options( wrapper->client, MYSQL_OPT_SSL_MODE, &val );

  return INT2NUM(result);
#endif
#ifdef NO_SSL_MODE_SUPPORT
  return Qnil;
#endif
}

/* call-seq:
 *    client.select_db(name)
 *
 * Causes the database specified by +name+ to become the default (current)
 * database on the connection specified by mysql.
 */
static VALUE rb_mysql_client_select_db(VALUE self, VALUE db)
{
  struct nogvl_select_db_args args;

  GET_CLIENT(self);
  REQUIRE_OPEN_DB(wrapper);

  args.mysql = wrapper->client;
  args.db = StringValuePtr(db);

  if (rb_thread_blocking_region(nogvl_select_db, &args, RUBY_UBF_IO, 0) == Qfalse)
    rb_raise_mysql2_error(wrapper); 

  return db;
}

/* call-seq:
 *    client.select_db(name)
 *
 * Causes the database specified by +name+ to become the default (current)
 * database on the connection specified by mysql.
 */
static VALUE rb_mysql_client_select_db(VALUE self, VALUE db)
{
  struct nogvl_select_db_args args;

  GET_CLIENT(self);
  REQUIRE_CONNECTED(wrapper);

  args.mysql = wrapper->client;
  args.db = StringValueCStr(db);

  if (rb_thread_call_without_gvl(nogvl_select_db, &args, RUBY_UBF_IO, 0) == Qfalse)
    rb_raise_mysql2_error(wrapper);

  return db;
}

static VALUE set_connect_timeout(VALUE self, VALUE value) {
  unsigned int connect_timeout = 0;
  GET_CLIENT(self);

  if(!NIL_P(value)) {
    connect_timeout = NUM2INT(value);
    if(0 == connect_timeout) return value;

    /* set default connection timeout behavior */
    if (mysql_options(wrapper->client, MYSQL_OPT_CONNECT_TIMEOUT, &connect_timeout)) {
      /* TODO: warning - unable to set connection timeout */
      rb_warn("%s\n", mysql_error(wrapper->client));
    }
  }
  return value;
}

static VALUE rb_mysql_client_socket(VALUE self) {
    GET_CLIENT(self);
    REQUIRE_OPEN_DB(wrapper);
    int fd_set_fd = wrapper->client->net.fd;
#ifdef _WIN32
    WSAPROTOCOL_INFO wsa_pi;
    // dupicate the SOCKET from libmysql
    int r = WSADuplicateSocket(wrapper->client->net.fd, GetCurrentProcessId(), &wsa_pi);
    SOCKET s = WSASocket(wsa_pi.iAddressFamily, wsa_pi.iSocketType, wsa_pi.iProtocol, &wsa_pi, 0, 0);
    // create the CRT fd so ruby can get back to the SOCKET
    fd_set_fd = _open_osfhandle(s, O_RDWR|O_BINARY);
    return INT2NUM(fd_set_fd);
#else
    return INT2NUM(fd_set_fd);
#endif
}

static VALUE disconnect_and_raise(VALUE self, VALUE error) {
  GET_CLIENT(self);

  MARK_CONN_INACTIVE(self);
  wrapper->connected = 0;

  /* Invalidate the MySQL socket to prevent further communication.
   * The GC will come along later and call mysql_close to free it.
   */
  if (invalidate_fd(wrapper->client->net.fd) == Qfalse) {
    fprintf(stderr, "[WARN] mysql2 failed to invalidate FD safely, closing unsafely\n");
    close(wrapper->client->net.fd);
  }

  rb_exc_raise(error);
}