Merge pull request #591 from justincase/prepared_mysql2_rebase

Prepared statements (rebase of #558, collected from #289)

Author: sodabrew

ext/mysql2/client.c

@@ -18,10 +18,11 @@ VALUE cMysql2Client;
 extern VALUE mMysql2, cMysql2Error;
 static VALUE sym_id, sym_version, sym_async, sym_symbolize_keys, sym_as, sym_array, sym_stream;
 static ID intern_merge, intern_merge_bang, intern_error_number_eql, intern_sql_state_eql;
+static ID intern_brackets, intern_new;
 
 #ifndef HAVE_RB_HASH_DUP
-static VALUE rb_hash_dup(VALUE other) {
-  return rb_funcall(rb_cHash, rb_intern("[]"), 1, other);
+VALUE rb_hash_dup(VALUE other) {
+  return rb_funcall(rb_cHash, intern_brackets, 1, other);
 }
 #endif
 
@@ -30,25 +31,12 @@ static VALUE rb_hash_dup(VALUE other) {
     rb_raise(cMysql2Error, "MySQL client is not initialized"); \
   }
 
-#define REQUIRE_CONNECTED(wrapper) \
-  REQUIRE_INITIALIZED(wrapper) \
-  if (!wrapper->connected && !wrapper->reconnect_enabled) { \
-    rb_raise(cMysql2Error, "closed MySQL connection"); \
-  }
-
 #define REQUIRE_NOT_CONNECTED(wrapper) \
   REQUIRE_INITIALIZED(wrapper) \
   if (wrapper->connected) { \
     rb_raise(cMysql2Error, "MySQL connection is already open"); \
   }
 
-#define MARK_CONN_INACTIVE(conn) \
-  wrapper->active_thread = Qnil;
-
-#define GET_CLIENT(self) \
-  mysql_client_wrapper *wrapper; \
-  Data_Get_Struct(self, mysql_client_wrapper, wrapper)
-
 /*
  * compatability with mysql-connector-c, where LIBMYSQL_VERSION is the correct
  * variable to use, but MYSQL_SERVER_VERSION gives the correct numbers when
@@ -136,7 +124,7 @@ static VALUE rb_raise_mysql2_error(mysql_client_wrapper *wrapper) {
   rb_enc_associate(rb_sql_state, rb_usascii_encoding());
 #endif
 
-  e = rb_funcall(cMysql2Error, rb_intern("new"), 2, rb_error_msg, LONG2FIX(wrapper->server_version));
+  e = rb_funcall(cMysql2Error, intern_new, 2, rb_error_msg, LONG2FIX(wrapper->server_version));
   rb_funcall(e, intern_error_number_eql, 1, UINT2NUM(mysql_errno(wrapper->client)));
   rb_funcall(e, intern_sql_state_eql, 1, rb_sql_state);
   rb_exc_raise(e);
@@ -247,6 +235,7 @@ static void rb_mysql_client_free(void *ptr) {
 void decr_mysql2_client(mysql_client_wrapper *wrapper)
 {
   wrapper->refcount--;
+
   if (wrapper->refcount == 0) {
     nogvl_close(wrapper);
     xfree(wrapper->client);
@@ -267,6 +256,7 @@ static VALUE allocate(VALUE klass) {
   wrapper->initialized = 0; /* means that that the wrapper is initialized */
   wrapper->refcount = 1;
   wrapper->client = (MYSQL*)xmalloc(sizeof(MYSQL));
+
   return obj;
 }
 
@@ -415,7 +405,7 @@ static VALUE do_send_query(void *args) {
   mysql_client_wrapper *wrapper = query_args->wrapper;
   if ((VALUE)rb_thread_call_without_gvl(nogvl_send_query, args, RUBY_UBF_IO, 0) == Qfalse) {
     /* an error occurred, we're not active anymore */
-    MARK_CONN_INACTIVE(self);
+    wrapper->active_thread = Qnil;
     return rb_raise_mysql2_error(wrapper);
   }
   return Qnil;
@@ -501,7 +491,7 @@ static VALUE rb_mysql_client_async_result(VALUE self) {
   current = rb_hash_dup(rb_iv_get(self, "@current_query_options"));
   RB_GC_GUARD(current);
   Check_Type(current, T_HASH);
-  resultObj = rb_mysql_result_to_obj(self, wrapper->encoding, current, result);
+  resultObj = rb_mysql_result_to_obj(self, wrapper->encoding, current, result, NULL);
 
   return resultObj;
 }
@@ -598,6 +588,25 @@ static VALUE finish_and_mark_inactive(void *args) {
 }
 #endif
 
+void rb_mysql_client_set_active_thread(VALUE self) {
+  VALUE thread_current = rb_thread_current();
+  GET_CLIENT(self);
+
+  // see if this connection is still waiting on a result from a previous query
+  if (NIL_P(wrapper->active_thread)) {
+    // mark this connection active
+    wrapper->active_thread = thread_current;
+  } else if (wrapper->active_thread == thread_current) {
+    rb_raise(cMysql2Error, "This connection is still waiting for a result, try again once you have the result");
+  } else {
+    VALUE inspect = rb_inspect(wrapper->active_thread);
+    const char *thr = StringValueCStr(inspect);
+
+    rb_raise(cMysql2Error, "This connection is in use by: %s", thr);
+    RB_GC_GUARD(inspect);
+  }
+}
+
 /* call-seq:
  *    client.abandon_results!
  *
@@ -641,7 +650,6 @@ static VALUE rb_mysql_client_query(int argc, VALUE * argv, VALUE self) {
   struct nogvl_send_query_args args;
   int async = 0;
   VALUE opts, current;
-  VALUE thread_current = rb_thread_current();
 #ifdef HAVE_RUBY_ENCODING_H
   rb_encoding *conn_enc;
 #endif
@@ -671,23 +679,10 @@ static VALUE rb_mysql_client_query(int argc, VALUE * argv, VALUE self) {
 #endif
   args.sql_ptr = StringValuePtr(args.sql);
   args.sql_len = RSTRING_LEN(args.sql);
-
-  /* see if this connection is still waiting on a result from a previous query */
-  if (NIL_P(wrapper->active_thread)) {
-    /* mark this connection active */
-    wrapper->active_thread = thread_current;
-  } else if (wrapper->active_thread == thread_current) {
-    rb_raise(cMysql2Error, "This connection is still waiting for a result, try again once you have the result");
-  } else {
-    VALUE inspect = rb_inspect(wrapper->active_thread);
-    const char *thr = StringValueCStr(inspect);
-
-    rb_raise(cMysql2Error, "This connection is in use by: %s", thr);
-    RB_GC_GUARD(inspect);
-  }
-
   args.wrapper = wrapper;
 
+  rb_mysql_client_set_active_thread(self);
+
 #ifndef _WIN32
   rb_rescue2(do_send_query, (VALUE)&args, disconnect_and_raise, self, rb_eException, (VALUE)0);
 
@@ -1080,7 +1075,7 @@ static VALUE rb_mysql_client_store_result(VALUE self)
   current = rb_hash_dup(rb_iv_get(self, "@current_query_options"));
   RB_GC_GUARD(current);
   Check_Type(current, T_HASH);
-  resultObj = rb_mysql_result_to_obj(self, wrapper->encoding, current, result);
+  resultObj = rb_mysql_result_to_obj(self, wrapper->encoding, current, result, NULL);
 
   return resultObj;
 }
@@ -1220,6 +1215,17 @@ static VALUE initialize_ext(VALUE self) {
   return self;
 }
 
+/* call-seq: client.prepare # => Mysql2::Statement
+ *
+ * Create a new prepared statement.
+ */
+static VALUE rb_mysql_client_prepare_statement(VALUE self, VALUE sql) {
+  GET_CLIENT(self);
+  REQUIRE_CONNECTED(wrapper);
+
+  return rb_mysql_stmt_new(self, sql);
+}
+
 void init_mysql2_client() {
   /* verify the libmysql we're about to use was the version we were built against
      https://github.com/luislavena/mysql-gem/commit/a600a9c459597da0712f70f43736e24b484f8a99 */
@@ -1265,6 +1271,7 @@ void init_mysql2_client() {
   rb_define_method(cMysql2Client, "async_result", rb_mysql_client_async_result, 0);
   rb_define_method(cMysql2Client, "last_id", rb_mysql_client_last_id, 0);
   rb_define_method(cMysql2Client, "affected_rows", rb_mysql_client_affected_rows, 0);
+  rb_define_method(cMysql2Client, "prepare", rb_mysql_client_prepare_statement, 1);
   rb_define_method(cMysql2Client, "thread_id", rb_mysql_client_thread_id, 0);
   rb_define_method(cMysql2Client, "ping", rb_mysql_client_ping, 0);
   rb_define_method(cMysql2Client, "select_db", rb_mysql_client_select_db, 1);
@@ -1299,6 +1306,8 @@ void init_mysql2_client() {
   sym_array           = ID2SYM(rb_intern("array"));
   sym_stream          = ID2SYM(rb_intern("stream"));
 
+  intern_brackets = rb_intern("[]");
+  intern_new = rb_intern("new");
   intern_merge = rb_intern("merge");
   intern_merge_bang = rb_intern("merge!");
   intern_error_number_eql = rb_intern("error_number=");

ext/mysql2/client.h

@@ -50,7 +50,28 @@ typedef struct {
   MYSQL *client;
 } mysql_client_wrapper;
 
+#define REQUIRE_CONNECTED(wrapper) \
+  REQUIRE_INITIALIZED(wrapper) \
+  if (!wrapper->connected && !wrapper->reconnect_enabled) { \
+    rb_raise(cMysql2Error, "closed MySQL connection"); \
+  }
+
+void rb_mysql_client_set_active_thread(VALUE self);
+
+#define MARK_CONN_INACTIVE(conn) do {\
+    GET_CLIENT(conn); \
+    wrapper->active_thread = Qnil; \
+  } while(0)
+
+#define GET_CLIENT(self) \
+  mysql_client_wrapper *wrapper; \
+  Data_Get_Struct(self, mysql_client_wrapper, wrapper);
+
 void init_mysql2_client();
 void decr_mysql2_client(mysql_client_wrapper *wrapper);
 
 #endif
+
+#ifndef HAVE_RB_HASH_DUP
+VALUE rb_hash_dup(VALUE other);
+#endif

ext/mysql2/mysql2_ext.c

@@ -9,4 +9,5 @@ void Init_mysql2() {
 
   init_mysql2_client();
   init_mysql2_result();
+  init_mysql2_statement();
 }

ext/mysql2/mysql2_ext.h

@@ -40,6 +40,7 @@ typedef unsigned int    uint;
 
 #include <client.h>
 #include <result.h>
+#include <statement.h>
 #include <infile.h>
 
 #endif