Require a result cursor on all prepared statements

Statement execute must return a cursor because Ruby is highly garbage
collected. If the connection is in results-streaming-mode for Statement A,
and in the middle Statement B gets garbage collected, a message will be
sent to the server notifying it to release Statement B, resulting in one
of these errors:
  Commands out of sync; you can't run this command now
  Row retrieval was canceled by mysql_stmt_close

By telling the server to give us a cursor to the result set, other
commands can be sent on the wire during results streaming.

Fixes #956

ext/mysql2/statement.c

@@ -117,7 +117,21 @@ VALUE rb_mysql_stmt_new(VALUE rb_client, VALUE sql) {
   {
     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");
+      rb_raise(cMysql2Error, "Unable to initialize prepared statement STMT_ATTR_UPDATE_MAX_LENGTH");
+    }
+  }
+
+  // Statement execute must return a cursor because Ruby is highly garbage
+  // collected. If the connection is in results-streaming-mode for Statement A,
+  // and in the middle Statement B gets garbage collected, a message will be
+  // sent to the server notifying it to release Statement B, resulting in one
+  // of these errors:
+  //   Commands out of sync; you can't run this command now
+  //   Row retrieval was canceled by mysql_stmt_close
+  {
+    unsigned long type = CURSOR_TYPE_READ_ONLY;
+    if (mysql_stmt_attr_set(stmt_wrapper->stmt, STMT_ATTR_CURSOR_TYPE, &type)) {
+      rb_raise(cMysql2Error, "Unable to initialize prepared statement CURSOR_TYPE_READ_ONLY");
     }
   }
 
@@ -403,24 +417,6 @@ static VALUE rb_mysql_stmt_execute(int argc, VALUE *argv, VALUE self) {
     }
   }
 
-  // Duplicate the options hash, merge! extra opts, put the copy into the Result object
-  current = rb_hash_dup(rb_iv_get(stmt_wrapper->client, "@query_options"));
-  (void)RB_GC_GUARD(current);
-  Check_Type(current, T_HASH);
-
-  // Merge in hash opts/keyword arguments
-  if (!NIL_P(opts)) {
-    rb_funcall(current, intern_merge_bang, 1, opts);
-  }
-
-  is_streaming = (Qtrue == rb_hash_aref(current, sym_stream));
-
-  // From stmt_execute to mysql_stmt_result_metadata to stmt_store_result, no
-  // Ruby API calls are allowed so that GC is not invoked. If the connection is
-  // in results-streaming-mode for Statement A, and in the middle Statement B
-  // gets garbage collected, a message will be sent to the server notifying it
-  // to release Statement B, resulting in a "Commands out of sync" error.
-
   if ((VALUE)rb_thread_call_without_gvl(nogvl_stmt_execute, stmt, RUBY_UBF_IO, 0) == Qfalse) {
     FREE_BINDS;
     rb_raise_mysql2_stmt_error(stmt_wrapper);
@@ -439,6 +435,17 @@ static VALUE rb_mysql_stmt_execute(int argc, VALUE *argv, VALUE self) {
     return Qnil;
   }
 
+  // Duplicate the options hash, merge! extra opts, put the copy into the Result object
+  current = rb_hash_dup(rb_iv_get(stmt_wrapper->client, "@query_options"));
+  (void)RB_GC_GUARD(current);
+  Check_Type(current, T_HASH);
+
+  // Merge in hash opts/keyword arguments
+  if (!NIL_P(opts)) {
+    rb_funcall(current, intern_merge_bang, 1, opts);
+  }
+
+  is_streaming = (Qtrue == rb_hash_aref(current, sym_stream));
   if (!is_streaming) {
     // recieve the whole result set from the server
     if (mysql_stmt_store_result(stmt)) {