-module(mysql_worker). -include_lib("kernel/include/logger.hrl"). -behaviour(gen_server). %% API -export([start_link/1]). -export([init/1, handle_call/3, handle_cast/2, handle_info/2, terminate/2, code_change/3]). -export([query/4, transaction/4]). -record(state, { name, conn }). -define(TIMEOUT, 5000). start_link(Args) -> gen_server:start_link(?MODULE, [Args], []). query(P, Query, Params, Timeout) when is_binary(Query) -> gen_server:call(P, {query, Query, Params, Timeout}, Timeout). transaction(P, Fun, Args, Timeout) -> gen_server:call(P, {transaction, Fun, Args}, Timeout). init([Args]) -> process_flag(trap_exit, true), Name = case lists:keyfind(name, 1, Args) of {name, {local, Name0}} -> Name0; _ -> ?MODULE end, ?LOG_INFO("~ts start...", [Name]), Args2 = lists:keydelete(name, 1, Args), {ok, Conn} = mysql:start_link(Args2), {ok, #state{name = Name, conn = Conn}}. handle_call({query, Query, Params, Timeout}, _From, #state{conn = Con} = State) -> Reply = handle_query(Con, Query, Params, Timeout), {reply, Reply, State}; handle_call({transaction, Fun, Args}, _From, #state{conn = Con} = State) -> Reply = handle_transaction(Con, Fun, Args), {reply, Reply, State}; handle_call(_Request, _From, State) -> {reply, ok, State}. handle_cast(_Msg, State) -> {noreply, State}. handle_info(_Info, State) -> {noreply, State}. terminate(_Reason, #state{name = Name, conn = Conn}) -> ?LOG_INFO("~ts stop...", [Name]), ok = mysql:stop(Conn), ok. code_change(_OldVsn, State, _Extra) -> {ok, State}. sql_type(SQL) when is_binary(SQL) -> L = string:to_upper(string:trim(erlang:binary_to_list(SQL))), % insert,select,update,delete case L of [$I, $N, $S, $E, $R, $T | _] -> {ok, insert}; [$S, $E, $L, $E, $C, $T | _] -> {ok, select}; [$U, $P, $D, $A, $T, $E | _] -> {ok, update}; [$D, $E, $L, $E, $T, $E | _] -> {ok, delete}; [$R, $E, $P, $L, $A, $C, $E | _] -> {ok, replace}; _ -> {error, unknown} end. handle_query(C, SQL, Params, Timeout) -> case mysql:query(C, SQL, Params, Timeout) of {ok, ColumnNames, Rows} -> {ok, ColumnNames, Rows}; ok -> M = #{affected_rows => mysql:affected_rows(C), warning_count => mysql:warning_count(C)}, case sql_type(SQL) of {ok, Type} when Type =:= insert orelse Type =:= replace -> LastInsertId = mysql:insert_id(C), {ok, M#{insert_id => LastInsertId}}; _ -> {ok, M} end; {error, Err} -> {error, format_error(Err)} end. handle_transaction(C, Fun, Args) -> mysql:transaction(C, fun() -> erlang:apply(Fun, [C | Args]) end). format_error({Code, State, Msg}) -> #{code => Code, sql_state => State, msg => Msg}.