support 'COM_RESET_CONNECTION' (#138)

A more lightweight version of COM_CHANGE_USER that does about the same to clean up the session state, but:

1. it does not re-authenticate (and do the extra client/server exchange for that)
2. it does not close the connection

More:
* https://dev.mysql.com/doc/dev/mysql-server/8.0.11/page_protocol_com_reset_connection.html
* https://dev.mysql.com/doc/refman/8.0/en/mysql-reset-connection.html

include/protocol.hrl

@@ -101,6 +101,7 @@
 -define(COM_STMT_RESET, 16#1a).
 -define(COM_SET_OPTION, 16#1b).
 -define(COM_STMT_FETCH, 16#1c).
+-define(COM_RESET_CONNECTION, 16#1f).
 
 %% --- Types ---
 

src/mysql.erl

@@ -32,7 +32,7 @@
          warning_count/1, affected_rows/1, autocommit/1, insert_id/1,
          encode/2, in_transaction/1,
          transaction/2, transaction/3, transaction/4,
-         change_user/3, change_user/4]).
+         change_user/3, change_user/4, reset_connection/1]).
 
 -export_type([connection/0, server_reason/0, query_result/0]).
 
@@ -153,7 +153,7 @@ start_link(Options) ->
         ServerName ->
             gen_server:start_link(ServerName, mysql_conn, Options, GenSrvOpts)
     end.
- 
+
 %% @see stop/2.
 -spec stop(Conn) -> ok
     when Conn :: connection().
@@ -400,7 +400,7 @@ execute(Conn, StatementRef, Params, FilterMap) when FilterMap == no_filtermap_fu
     execute(Conn, StatementRef, Params, FilterMap, default_timeout).
 
 %% @doc Executes a prepared statement.
-%% 
+%%
 %% The `FilterMap' and `Timeout' arguments are optional.
 %% <ul>
 %%   <li>If the `FilterMap' argument is the atom `no_filtermap_fun' or is
@@ -620,6 +620,12 @@ execute_transaction(Conn, Fun, Args, Retries) ->
             %% an error.
             ok = gen_server:call(Conn, rollback, infinity),
             erlang:raise(error, E, ?GET_STACK(Stacktrace));
+        ?EXCEPTION(error, reset_connection_in_transaction = E, Stacktrace) ->
+            %% The called tried to reset connection inside the transaction, which
+            %% is not allowed and a serious mistake. We roll back and raise
+            %% an error.
+            ok = gen_server:call(Conn, rollback, infinity),
+            erlang:raise(error, E, ?GET_STACK(Stacktrace));
         ?EXCEPTION(Class, Reason, Stacktrace) ->
             %% We must be able to rollback. Otherwise let's crash.
             ok = gen_server:call(Conn, rollback, infinity),
@@ -681,6 +687,16 @@ change_user(Conn, Username, Password, Options) ->
     end,
     gen_server:call(Conn, {change_user, Username, Password, Options}).
 
+-spec reset_connection(Conn) -> ok | {error, Reason}
+    when Conn :: connection(),
+         Reason :: server_reason().
+reset_connection(Conn) ->
+    case in_transaction(Conn) of
+        true -> error(reset_connection_in_transaction);
+        false -> ok
+    end,
+    gen_server:call(Conn, reset_connection).
+
 %% @doc Encodes a term as a MySQL literal so that it can be used to inside a
 %% query. If backslash escapes are enabled, backslashes and single quotes in
 %% strings and binaries are escaped. Otherwise only single quotes are escaped.

src/mysql_conn.erl

@@ -291,7 +291,7 @@ handle_call({change_user, Username, Password, Options}, From,
     Prepares = proplists:get_value(prepare, Options, []),
     setopts(SockMod, Socket, [{active, false}]),
     Result = mysql_protocol:change_user(SockMod, Socket, Username, Password,
-                                        AuthPluginData, Database, 
+                                        AuthPluginData, Database,
                                         ServerVersion),
     setopts(SockMod, Socket, [{active, once}]),
     State1 = update_state(Result, State),
@@ -312,6 +312,20 @@ handle_call({change_user, Username, Password, Options}, From,
             gen_server:reply(From, {error, error_to_reason(E)}),
             stop_server(change_user_failed, State2)
     end;
+handle_call(reset_connection, _From, #state{socket = Socket, sockmod = SockMod} = State) ->
+    setopts(SockMod, Socket, [{active, false}]),
+    Result = mysql_protocol:reset_connnection(SockMod, Socket),
+    setopts(SockMod, Socket, [{active, once}]),
+    State1 = update_state(Result, State),
+    Reply = case Result of
+        #ok{} -> ok;
+        #error{} = E ->
+            %% 'COM_RESET_CONNECTION' is added in MySQL 5.7 and MariaDB 10
+            %% "Unkown command" is returned when MySQL =< 5.6 or MariaDB =< 5.5
+            {error, error_to_reason(E)}
+    end,
+    {reply, Reply, State1};
+
 handle_call(warning_count, _From, State) ->
     {reply, State#state.warning_count, State};
 handle_call(insert_id, _From, State) ->

src/mysql_protocol.erl

@@ -31,7 +31,8 @@
          query/4, query/5, fetch_query_response/3,
          fetch_query_response/4, prepare/3, unprepare/3,
          execute/5, execute/6, fetch_execute_response/3,
-         fetch_execute_response/4, valid_params/1]).
+         fetch_execute_response/4, reset_connnection/2,
+	       valid_params/1]).
 
 -type query_filtermap() :: no_filtermap_fun
                          | fun(([term()]) -> query_filtermap_res())
@@ -234,7 +235,7 @@ fetch_execute_response(SockModule, Socket, FilterMap, Timeout) ->
 %% @doc Changes the user of the connection.
 -spec change_user(module(), term(), iodata(), iodata(), binary(),
                   undefined | iodata(), [integer()]) -> #ok{} | #error{}.
-change_user(SockModule, Socket, Username, Password, Salt, Database, 
+change_user(SockModule, Socket, Username, Password, Salt, Database,
             ServerVersion) ->
     DbBin = case Database of
         undefined -> <<>>;
@@ -253,6 +254,17 @@ change_user(SockModule, Socket, Username, Password, Salt, Database,
             parse_error_packet(Packet)
     end.
 
+-spec reset_connnection(module(), term()) -> #ok{}|#error{}.
+reset_connnection(SockModule, Socket) ->
+    {ok, SeqNum1} = send_packet(SockModule, Socket, <<?COM_RESET_CONNECTION>>, 0),
+    {ok, Packet, _SeqNum2} = recv_packet(SockModule, Socket, SeqNum1),
+    case Packet of
+        ?ok_pattern ->
+            parse_ok_packet(Packet);
+        ?error_pattern ->
+            parse_error_packet(Packet)
+    end.
+
 %% --- internal ---
 
 %% @doc Parses a handshake. This is the first thing that comes from the server

test/mysql_tests.erl

@@ -180,6 +180,25 @@ keep_alive_test() ->
      ?assertMatch({crash_report, _}, LoggedReport),
      ?assertExit(noproc, mysql:stop(Pid)).
 
+reset_connection_test() ->
+    %% Ignored test with MySQL earlier than 5.7
+    Options = [{user, ?user}, {password, ?password}, {keep_alive, true}],
+    {ok, Pid} = mysql:start_link(Options),
+    ok = mysql:query(Pid, <<"CREATE DATABASE otptest">>),
+    ok = mysql:query(Pid, <<"USE otptest">>),
+    ok = mysql:query(Pid, <<"SET autocommit = 1">>),
+    ok = mysql:query(Pid, ?create_table_t),
+    ok = mysql:query(Pid, <<"INSERT INTO t (id, tx) VALUES (1, 'text 1')">>),
+    ?assertEqual(1, mysql:insert_id(Pid)),  %% auto_increment starts from 1
+    case mysql:reset_connection(Pid) of
+      ok ->
+        ?assertEqual(0, mysql:insert_id(Pid)); %% insertid reset to 0;
+      _Error ->
+        ?assertEqual(1, mysql:insert_id(Pid)) %% reset failed
+    end,
+    mysql:stop(Pid),
+    ok.
+
 unix_socket_test() ->
     try
         list_to_integer(erlang:system_info(otp_release))
@@ -212,7 +231,7 @@ unix_socket_test() ->
             error_logger:info_msg("Skipping unix socket tests. Current OTP "
                                   "release could not be determined.~n")
     end.
-    
+
 connect_queries_failure_test() ->
     process_flag(trap_exit, true),
     {error, Reason} = mysql:start_link([{user, ?user}, {password, ?password},