emysql_query and emysql_saver

.gitignore

@@ -23,3 +23,4 @@ test/ct_log_cache
 logs/*
 .#*
 include/crypto_compat.hrl
+deps/*

README.md

@@ -284,6 +284,18 @@ or (after building emysql.app and the database, as explained above), start a_hel
 
 General Notes on using Emysql, including the actual specs:
 
+#### Resolve Emysql dependency via rebar.config if possible
+
+```Erlang
+{deps,
+[ {'emysql', ".*", { git, "git://github.com/Eonblast/Emysql.git",   "master" } } ]
+```
+or some other branch
+```Erlang
+{deps,
+[ {'emysql', ".*", { git, "git://github.com/jacktang/Emysql.git",   "taodi" } } ]
+```
+
 #### Starting an Application
 
 The Emysql driver is an Erlang gen-server, and, application.

include/emysql.hrl

@@ -34,9 +34,9 @@
 	       port :: number(), 
 	       database :: string(), 
 	       encoding :: utf8 | latin1 | {utf8, utf8_unicode_ci} | {utf8, utf8_general_ci},
-	       available=queue:new() :: queue(), 
-	       locked=gb_trees:empty() :: gb_tree(), 
-	       waiting=queue:new() :: queue(), 
+	       available=queue:new() :: queue:queue(), 
+	       locked=gb_trees:empty() :: gb_tree:tree(), 
+	       waiting=queue:new() :: queue:queue(), 
 	       start_cmds=[] :: string(), 
 	       conn_test_period=0 :: number(), 
 	       connect_timeout=infinity :: number() | infinity,
@@ -180,3 +180,8 @@
 %  number of result set columns.
 -define(SERVER_STATUS_METADATA_CHANGED, 1024).
 
+%  Wrap the query result in erlang record
+-define(AS_REC(RecAtom), [RecAtom, record_info(fields, RecAtom)]).
+-define(AS_VAL,       as_val).
+% 
+-define(INPUT(RecAtom, Records), [Records, record_info(fields, RecAtom)]).

rebar.config

@@ -1,9 +1,20 @@
 % -*- Erlang -*-
 % vim: ts=4 sw=4 et ft=erlang
 {erl_opts, [
-        nowarn_deprecated_type
+        nowarn_deprecated_type,
+        {parse_transform, lager_transform}    
 ]}.
+
 {pre_hooks,[
         {"linux|bsd|darwin|solaris", compile, "escript ./support/crypto_compat.escript"},
         {"win32", compile, "escript.exe support/crypto_compat.escript"}
 ]}.
+
+{deps, [
+        {'espec',     ".*",   { git, "git://github.com/lucaspiller/espec.git",  "master"  } },
+        {'lager',     ".*",   { git, "git://github.com/basho/lager.git",        "master"  } },
+        {'erl_utils', ".*",   { git, "git://github.com/jacktang/erl_utils.git", "master"  } }
+]}.
+
+
+{xref_warnings, true}.

src/emysql.app.src

@@ -10,6 +10,11 @@
     {registered, [emysql_conn_mgr, emysql_sup]},
     {applications, [kernel, stdlib, crypto]},
     {env, [
-      {default_timeout, 5000},
-      {conn_test_period, 28000}]}
+           {default_timeout,  5000},
+           {conn_test_period, 28000},
+           {default_conn_opt,  [ {host, "localhost"},
+                                 {user, "root"},
+                                 {database, test}
+                                ]}
+          ]}
 ]}.

src/emysql.erl

@@ -114,6 +114,8 @@
             default_timeout/0
 ]).
 
+-export([transaction/2, transaction/3, abort/1]).
+
 %% Result Conversion API
 -export([
          as_dict/1,
@@ -631,6 +633,16 @@ execute(PoolId, StmtName, Args, Timeout, nonblocking) when is_atom(StmtName), is
             unavailable
     end.
 
+transaction(PoolId, Fun) ->
+    transaction(PoolId, Fun, default_timeout()).
+
+transaction(PoolId, Fun, Timeout) ->
+    Connection = emysql_conn_mgr:wait_for_connection(PoolId),
+    monitor_work(Connection, Timeout, [Connection, transaction, Fun]).
+
+abort(Reason) ->
+    throw(Reason).
+
 %% @doc Return the field names of a result packet
 %% @end
 -spec field_names(Result) -> [Name]
@@ -669,7 +681,7 @@ result_type(#eof_packet{})    -> eof.
 -spec as_dict(Result) -> Dict
   when
     Result :: #result_packet{},
-    Dict :: dict().
+    Dict :: dict:dict().
 as_dict(Res) -> emysql_conv:as_dict(Res).
 
 

src/emysql_conn.erl

@@ -86,13 +86,37 @@ set_encoding(Connection, Encoding) ->
 canonicalize_query(Q) when is_binary(Q) -> Q;
 canonicalize_query(QL) when is_list(QL) -> iolist_to_binary(QL).
 
+execute(Connection, transaction, Fun) when is_function(Fun) ->
+    case begin_transaction(Connection) of
+        #ok_packet{} ->
+            try Fun(Connection) of
+                Val ->
+                    case commit_transaction(Connection) of
+                        #ok_packet{} ->
+                            {atomic, Val};
+                        #error_packet{} = ErrorPacket ->
+                            {aborted, {commit_error, ErrorPacket}}
+                    end
+            catch
+                throw:Reason ->
+                    rollback_transaction(Connection),
+                    {aborted, Reason};
+                Class:Exception ->
+                    rollback_transaction(Connection),
+                    erlang:raise(Class, Exception, erlang:get_stacktrace())
+            end;
+        #error_packet{} = ErrorPacket ->
+            {aborted, {begin_error, ErrorPacket}}
+    end;
 execute(Connection, StmtName, []) when is_atom(StmtName) ->
     prepare_statement(Connection, StmtName),
     StmtNameBin = atom_to_binary(StmtName, utf8),
+    execute_trace:execute(StmtName),
     Packet = <<?COM_QUERY, "EXECUTE ", StmtNameBin/binary>>,
     send_recv(Connection, Packet);
 execute(Connection, Query, []) ->
     QB = canonicalize_query(Query),
+    execute_trace:execute(raw_query, QB),
     Packet = <<?COM_QUERY, QB/binary>>,
     send_recv(Connection, Packet);
 execute(Connection, Query, Args) when (is_list(Query) orelse is_binary(Query)) andalso is_list(Args) ->
@@ -103,6 +127,7 @@ execute(Connection, Query, Args) when (is_list(Query) orelse is_binary(Query)) a
         OK when is_record(OK, ok_packet) ->
             ParamNamesBin = list_to_binary(string:join([[$@ | integer_to_list(I)] || I <- lists:seq(1, length(Args))], ", ")),  % todo: utf8?
             Packet = <<?COM_QUERY, "EXECUTE ", (list_to_binary(StmtName))/binary, " USING ", ParamNamesBin/binary>>,  % todo: utf8?
+            execute_trace:execute(StmtName, Args),
             send_recv(Connection, Packet);
         Error ->
             Error
@@ -126,6 +151,7 @@ prepare(Connection, Name, Statement) when is_atom(Name) ->
     prepare(Connection, atom_to_list(Name), Statement);
 prepare(Connection, Name, Statement) ->
     StatementBin = encode(Statement, binary),
+    execute_trace:prepare(Name, Statement),
     Packet = <<?COM_QUERY, "PREPARE ", (list_to_binary(Name))/binary, " FROM ", StatementBin/binary>>,  % todo: utf8?
     case send_recv(Connection, Packet) of
         OK when is_record(OK, ok_packet) ->
@@ -137,9 +163,22 @@ prepare(Connection, Name, Statement) ->
 unprepare(Connection, Name) when is_atom(Name)->
     unprepare(Connection, atom_to_list(Name));
 unprepare(Connection, Name) ->
+    execute_trace:unprepare(Name),
     Packet = <<?COM_QUERY, "DEALLOCATE PREPARE ", (list_to_binary(Name))/binary>>,  % todo: utf8?
     send_recv(Connection, Packet).
 
+begin_transaction(Connection) ->
+    execute_trace:begin_transaction(),
+    emysql_conn:execute(Connection, <<"BEGIN">>, []).
+
+rollback_transaction(Connection) ->
+    execute_trace:rollback_transaction(),
+    emysql_conn:execute(Connection, <<"ROLLBACK">>, []).
+
+commit_transaction(Connection) ->
+    execute_trace:commit_transaction(),
+    emysql_conn:execute(Connection, <<"COMMIT">>, []).
+
 open_n_connections(PoolId, N) ->
     case emysql_conn_mgr:find_pool(PoolId, emysql_conn_mgr:pools()) of
         {Pool, _} ->

src/emysql_conn_mgr.erl

@@ -40,7 +40,7 @@
 
 -include("emysql.hrl").
 
--record(state, {pools, lockers = dict:new() :: dict()}).
+-record(state, {pools, lockers = dict:new() :: dict:dict()}).
 
 %%====================================================================
 %% API

src/emysql_conv.erl

@@ -11,6 +11,7 @@
          as_json/1,
          as_proplist/1,
          as_record/3,
+         as_record/5,
          as_record/4
 ]).
 
@@ -28,15 +29,15 @@ as_proplist(#result_packet{field_list=_Cols,rows=_Rows}) when is_list(_Cols),
 as_proplist(#result_packet{field_list=_Cols,rows=_Rows}) when is_list(_Cols),
                                                               _Rows =:= [] ->
     [];
-as_proplist(Res = #result_packet{field_list=Cols,rows=Rows}) when is_list(Cols),
+as_proplist(#result_packet{field_list=Cols,rows=Rows} = Res) when is_list(Cols),
                                                                   is_list(Rows) ->
     Fields = emysql:field_names(Res),
     [begin
         [{K, V} || {K, V} <- lists:zip(Fields, R)]
     end || R <- Rows].
 
 %% @see emysql:as_record/1
-as_record(Result = #result_packet{}, RecordName, Fields, Fun) when is_atom(RecordName), is_list(Fields), is_function(Fun) ->
+as_record(#result_packet{} = Result, RecordName, Fields, Fun, AccIn) when is_atom(RecordName), is_list(Fields), is_function(Fun) ->
     Columns = Result#result_packet.field_list,
 
     S = lists:seq(1, length(Columns)),
@@ -50,13 +51,28 @@ as_record(Result = #result_packet{}, RecordName, Fields, Fun) when is_atom(Recor
         end
     end,
     Fs = [ F(FieldName) || FieldName <- Fields ],
-    F1 = fun(Row) ->
-        RecordData = [ Fx(Row) || Fx <- Fs ],
-        Fun(list_to_tuple([RecordName|RecordData]))
-    end,
-    [ F1(Row) || Row <- Result#result_packet.rows ].
 
-as_record(Result = #result_packet{}, RecordName, Fields) when is_atom(RecordName), is_list(Fields) ->
+    case AccIn of
+        undefined ->
+            F1 = fun(Row) ->
+                         RecordData = [ Fx(Row) || Fx <- Fs ],
+                         Fun(list_to_tuple([RecordName|RecordData]))
+                 end,
+            [ F1(Row) || Row <- Result#result_packet.rows ];
+        _ ->
+            lists:foldl(fun(Row, {RsAccIn, FunAccIn}) ->
+                                RecordData = [ Fx(Row) || Fx <- Fs ],
+                                Rec = list_to_tuple([RecordName|RecordData]),
+                                NFunAccIn = Fun(Rec, FunAccIn),
+                                {lists:append(RsAccIn, [Rec]), NFunAccIn}
+                        end, {[], AccIn}, Result#result_packet.rows)
+    end.
+
+as_record(#result_packet{} = Result, RecordName, Fields, Fun) when is_atom(RecordName), is_list(Fields), is_function(Fun) ->
+    as_record(Result, RecordName, Fields, Fun, undefined).
+
+
+as_record(#result_packet{} = Result, RecordName, Fields) when is_atom(RecordName), is_list(Fields) ->
     as_record(Result, RecordName, Fields, fun(A) -> A end).
 
 %% @see emysql:as_json/1