%-*-Mode:erlang;coding:utf-8;tab-width:4;c-basic-offset:4;indent-tabs-mode:()-*- % ex: set ft=erlang fenc=utf-8 sts=4 ts=4 sw=4 et: %%% %%%------------------------------------------------------------------------ %%% @doc %%% ==CloudI MySQL Data Module== %%% @end %%% %%% BSD LICENSE %%% %%% Copyright (c) 2009-2015, Michael Truog %%% All rights reserved. %%% %%% Redistribution and use in source and binary forms, with or without %%% modification, are permitted provided that the following conditions are met: %%% %%% * Redistributions of source code must retain the above copyright %%% notice, this list of conditions and the following disclaimer. %%% * Redistributions in binary form must reproduce the above copyright %%% notice, this list of conditions and the following disclaimer in %%% the documentation and/or other materials provided with the %%% distribution. %%% * All advertising materials mentioning features or use of this %%% software must display the following acknowledgment: %%% This product includes software developed by Michael Truog %%% * The name of the author may not be used to endorse or promote %%% products derived from this software without specific prior %%% written permission %%% %%% THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND %%% CONTRIBUTORS "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, %%% INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES %%% OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE %%% DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR %%% CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, %%% SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, %%% BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR %%% SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS %%% INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, %%% WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING %%% NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE %%% OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH %%% DAMAGE. %%% %%% @author Michael Truog %%% @copyright 2009-2014 Michael Truog %%% @version 1.5.0 {@date} {@time} %%%------------------------------------------------------------------------ -module(cloudi_service_db_mysql). -author('mjtruog [at] gmail (dot) com'). -behaviour(cloudi_service). %% external interface -export([equery/4, equery/5, prepare_query/4, prepare_query/5, execute_query/4, execute_query/5, squery/3, squery/4, transaction/3, transaction/4]). %% cloudi_service callbacks -export([cloudi_service_init/4, cloudi_service_handle_request/11, cloudi_service_handle_info/3, cloudi_service_terminate/3]). -include_lib("cloudi_core/include/cloudi_logger.hrl"). -define(DEFAULT_DATABASE, undefined). % required argument, string -define(DEFAULT_DRIVER, eonblast). -define(DEFAULT_HOST_NAME, "127.0.0.1"). -define(DEFAULT_USER_NAME, "cloudi"). -define(DEFAULT_PASSWORD, ""). -define(DEFAULT_PORT, 3306). -define(DEFAULT_PING, undefined). % ms -define(DEFAULT_TIMEOUT, 20000). % ms (connect-only) -define(DEFAULT_ENCODING, utf8). -define(DEFAULT_ENDIAN, native). % response binary sizes -define(DEFAULT_OUTPUT, both). -define(DEFAULT_INTERNAL_INTERFACE, native). -define(DEFAULT_START_COMMANDS, []). % list(SQL :: binary()) -define(DEFAULT_DEBUG, false). % log output for debugging -define(DEFAULT_DEBUG_LEVEL, trace). % supported drivers -define(MODULE_EONBLAST, emysql_conn). % default -type endian() :: big | little | native. -record(state, { module :: ?MODULE_EONBLAST, connection :: any(), endian :: endian(), output :: both | external | internal, interface :: native | common, debug_level :: off | trace | debug | info | warn | error | fatal }). % required for ?MODULE_EONBLAST -record(pool, { pool_id, size, user, password, host, port, database, encoding, available = queue:new(), locked = gb_trees:empty(), waiting = queue:new(), start_cmds = [], conn_test_period = 0, connect_timeout = infinity }). -record(field, { seq_num :: pos_integer(), catalog :: binary(), db :: binary(), table :: binary(), org_table :: binary(), name :: binary(), org_name :: binary(), type :: non_neg_integer(), default :: binary(), charset_nr :: non_neg_integer(), length :: non_neg_integer(), flags :: non_neg_integer(), decimals :: non_neg_integer(), % if row data is already decoded, why provide the decoder? decoder :: fun((binary()) -> any()) }). -record(ok_packet, { seq_num :: pos_integer(), affected_rows :: non_neg_integer(), insert_id :: integer() | binary(), status :: non_neg_integer(), warning_count :: non_neg_integer(), msg :: list() | {error, list(), any()} | {incomplete, list(), binary()} }). -record(error_packet, { seq_num :: pos_integer(), code :: non_neg_integer(), status :: 0 | binary(), msg :: list() }). -record(eof_packet, { seq_num, status :: non_neg_integer(), warning_count :: non_neg_integer() }). -record(result_packet, { seq_num, field_list :: list(#field{}), rows :: list(list(any())), extra :: binary() }). %%%------------------------------------------------------------------------ %%% External interface functions %%%------------------------------------------------------------------------ -type count() :: non_neg_integer(). -type rows() :: list(tuple()). -type common_result() :: {updated, count()} | %{updated, count(), rows()} | {selected, rows()} | {error, any()}. -export_type([common_result/0]). -type agent() :: cloudi:agent(). -type service_name() :: cloudi:service_name(). -type timeout_milliseconds() :: cloudi:timeout_milliseconds(). -type external_response(Result) :: {{ok, Result}, NewAgent :: agent()} | {{error, cloudi:error_reason_sync()}, NewAgent :: agent()}. %%------------------------------------------------------------------------- %% @doc %% ===Perform an extended SQL query that does string substitution with '?'s.=== %% @end %%------------------------------------------------------------------------- -spec equery(Agent :: agent(), Name :: service_name(), Query :: string() | binary(), Parameters :: list()) -> external_response(any()). equery(Agent, Name, Query, Parameters) when (is_binary(Query) orelse is_list(Query)), is_list(Parameters) -> cloudi:send_sync(Agent, Name, {Query, Parameters}). %%------------------------------------------------------------------------- %% @doc %% ===Perform an extended SQL query that does string substitution with '?'s using a timeout.=== %% @end %%------------------------------------------------------------------------- -spec equery(Agent :: agent(), Name :: service_name(), Query :: string() | binary(), Parameters :: list(), Timeout :: timeout_milliseconds()) -> external_response(any()). equery(Agent, Name, Query, Parameters, Timeout) when (is_binary(Query) orelse is_list(Query)), is_list(Parameters) -> cloudi:send_sync(Agent, Name, {Query, Parameters}, Timeout). %%------------------------------------------------------------------------- %% @doc %% ===Prepare a query with '?'s for dynamic values.=== %% The query can be used in the future with the supplied Identifier atom. %% @end %%------------------------------------------------------------------------- -spec prepare_query(Agent :: agent(), Name :: service_name(), Identifier :: atom(), Query :: string() | binary()) -> external_response(ok). prepare_query(Agent, Name, Identifier, Query) when is_atom(Identifier), (is_binary(Query) orelse is_list(Query)) -> cloudi:send_sync(Agent, Name, {prepare, Identifier, Query}). %%------------------------------------------------------------------------- %% @doc %% ===Prepare a query with '?'s for dynamic values using a timeout.=== %% The query can be used in the future with the supplied Identifier atom. %% @end %%------------------------------------------------------------------------- -spec prepare_query(Agent :: agent(), Name :: service_name(), Identifier :: atom(), Query :: string() | binary(), Timeout :: timeout_milliseconds()) -> external_response(ok). prepare_query(Agent, Name, Identifier, Query, Timeout) when is_atom(Identifier), (is_binary(Query) orelse is_list(Query)) -> cloudi:send_sync(Agent, Name, {prepare, Identifier, Query}, Timeout). %%------------------------------------------------------------------------- %% @doc %% ===Execute a query that has already been prepared.=== %% @end %%------------------------------------------------------------------------- -spec execute_query(Agent :: agent(), Name :: service_name(), Identifier :: atom(), Arguments :: list()) -> external_response(any()). execute_query(Agent, Name, Identifier, Arguments) when is_atom(Identifier), is_list(Arguments) -> cloudi:send_sync(Agent, Name, {execute, Identifier, Arguments}). %%------------------------------------------------------------------------- %% @doc %% ===Execute a query that has already been prepared with a timeout.=== %% @end %%------------------------------------------------------------------------- -spec execute_query(Agent :: agent(), Name :: service_name(), Identifier :: atom(), Arguments :: list(), Timeout :: timeout_milliseconds()) -> external_response(any()). execute_query(Agent, Name, Identifier, Arguments, Timeout) when is_atom(Identifier), is_list(Arguments) -> cloudi:send_sync(Agent, Name, {execute, Identifier, Arguments}, Timeout). %%------------------------------------------------------------------------- %% @doc %% ===Perform a simple SQL string query.=== %% @end %%------------------------------------------------------------------------- -spec squery(Agent :: agent(), Name :: service_name(), Query :: string() | binary()) -> external_response(any()). squery(Agent, Name, Query) when (is_binary(Query) orelse is_list(Query)) -> cloudi:send_sync(Agent, Name, Query). %%------------------------------------------------------------------------- %% @doc %% ===Perform a simple SQL string query with a timeout.=== %% @end %%------------------------------------------------------------------------- -spec squery(Agent :: agent(), Name :: service_name(), Query :: string() | binary(), Timeout :: timeout_milliseconds()) -> external_response(any()). squery(Agent, Name, Query, Timeout) when (is_binary(Query) orelse is_list(Query)) -> cloudi:send_sync(Agent, Name, Query, Timeout). %%------------------------------------------------------------------------- %% @doc %% ===A list of SQL simple query for a transaction.=== %% @end %%------------------------------------------------------------------------- -spec transaction(Agent :: agent(), Name :: service_name(), QueryList :: list(string() | binary())) -> external_response(ok | {error, any()}). transaction(Agent, Name, [Query | _] = QueryList) when (is_binary(Query) orelse is_list(Query)) -> cloudi:send_sync(Agent, Name, QueryList). %%------------------------------------------------------------------------- %% @doc %% ===A list of SQL simple query for a transaction with a timeout.=== %% @end %%------------------------------------------------------------------------- -spec transaction(Agent :: agent(), Name :: service_name(), QueryList :: list(string() | binary()), Timeout :: timeout_milliseconds()) -> external_response(ok | {error, any()}). transaction(Agent, Name, [Query | _] = QueryList, Timeout) when (is_binary(Query) orelse is_list(Query)) -> cloudi:send_sync(Agent, Name, QueryList, Timeout). %%%------------------------------------------------------------------------ %%% Callback functions from cloudi_service %%%------------------------------------------------------------------------ cloudi_service_init(Args, _Prefix, _Timeout, Dispatcher) -> Defaults = [ {database, ?DEFAULT_DATABASE}, {driver, ?DEFAULT_DRIVER}, {hostname, ?DEFAULT_HOST_NAME}, {username, ?DEFAULT_USER_NAME}, {password, ?DEFAULT_PASSWORD}, {port, ?DEFAULT_PORT}, {ping, ?DEFAULT_PING}, {timeout, ?DEFAULT_TIMEOUT}, {encoding, ?DEFAULT_ENCODING}, {endian, ?DEFAULT_ENDIAN}, {output, ?DEFAULT_OUTPUT}, {internal_interface, ?DEFAULT_INTERNAL_INTERFACE}, {start_commands, ?DEFAULT_START_COMMANDS}, {debug, ?DEFAULT_DEBUG}, {debug_level, ?DEFAULT_DEBUG_LEVEL}], [Database, Driver, HostName, UserName, Password, Port, Ping, TimeoutConnect, Encoding, Endian, Output, Interface, StartCommands, Debug, DebugLevel] = cloudi_proplists:take_values(Defaults, Args), true = (is_list(Database) andalso is_integer(hd(Database))), Module = if Driver =:= eonblast -> ?MODULE_EONBLAST end, true = (is_list(HostName) andalso is_integer(hd(HostName))), true = (is_list(UserName) andalso is_integer(hd(UserName))), true = (is_list(Password) andalso is_integer(hd(Password))), true = (is_integer(Port) andalso (Port > 0)), TimeoutMax = cloudi_service:timeout_max(Dispatcher), true = (Ping =:= undefined) orelse (is_integer(Ping) andalso (Ping > 0) andalso (Ping =< TimeoutMax)), true = (is_integer(TimeoutConnect) andalso (TimeoutConnect > 0)), true = (Encoding =:= utf8) orelse (Encoding =:= latin1), true = ((Endian =:= big) orelse (Endian =:= little) orelse (Endian =:= native)), true = ((Output =:= both) orelse (Output =:= external) orelse (Output =:= internal)), true = ((Interface =:= native) orelse (Interface =:= common)), true = ((Debug =:= true) orelse (Debug =:= false)), true = ((DebugLevel =:= trace) orelse (DebugLevel =:= debug) orelse (DebugLevel =:= info) orelse (DebugLevel =:= warn) orelse (DebugLevel =:= error) orelse (DebugLevel =:= fatal)), case driver_open(Module, HostName, UserName, Password, Port, Database, TimeoutConnect, Encoding, StartCommands) of {ok, Connection} -> cloudi_service:subscribe(Dispatcher, Database), DebugLogLevel = if Debug =:= false -> off; Debug =:= true -> DebugLevel end, if Ping =:= undefined -> ok; is_integer(Ping) -> erlang:send_after(Ping, cloudi_service:self(Dispatcher), {ping, Ping}) end, {ok, #state{module = Module, connection = Connection, endian = Endian, output = Output, interface = Interface, debug_level = DebugLogLevel}}; {error, Reason} -> {stop, Reason} end. cloudi_service_handle_request(_Type, _Name, _Pattern, _RequestInfo, Request, Timeout, _Priority, _TransId, _Pid, #state{output = Output} = State, _Dispatcher) -> case Request of {Query, []} when (Output =/= external), (is_binary(Query) orelse is_list(Query)) -> Response = driver_squery(internal, Query, Timeout, State), {reply, Response, State}; {Query, Parameters} when (Output =/= external), (is_binary(Query) orelse is_list(Query)) -> Response = driver_equery(Query, Parameters, Timeout, State), {reply, Response, State}; [Query | _] = QueryList when (Output =/= external), (is_binary(Query) orelse is_list(Query)) -> Response = driver_with_transaction(QueryList, Timeout, State), {reply, Response, State}; Query when is_binary(Query) -> ResponseOutput = if Output =:= internal -> internal; Output =:= external; Output =:= both -> external end, Response = driver_squery(ResponseOutput, Query, Timeout, State), {reply, Response, State}; Query when (Output =/= external), is_list(Query) -> Response = driver_squery(internal, Query, Timeout, State), {reply, Response, State}; % older requests not meant to be compatible with other % CloudI database services {prepare, Identifier, Query} when (Output =/= external), (is_binary(Query) orelse is_list(Query)) -> Response = driver_prepare(Identifier, Query, Timeout, State), {reply, Response, State}; {execute, Identifier, Parameters} when (Output =/= external) -> Response = driver_equery(Identifier, Parameters, Timeout, State), {reply, Response, State}; {equery, Query, []} when (Output =/= external), (is_binary(Query) orelse is_list(Query)) -> Response = driver_squery(internal, Query, Timeout, State), {reply, Response, State}; {equery, Query, Parameters} when (Output =/= external), (is_binary(Query) orelse is_list(Query)) -> Response = driver_equery(Query, Parameters, Timeout, State), {reply, Response, State}; {squery, Query} when (Output =/= external), (is_binary(Query) orelse is_list(Query)) -> Response = driver_squery(internal, Query, Timeout, State), {reply, Response, State} end. cloudi_service_handle_info({ping, Ping} = Request, State, Dispatcher) -> case driver_ping(Ping, State) of ok -> erlang:send_after(Ping, cloudi_service:self(Dispatcher), Request), {noreply, State}; error -> {stop, ping_failed, State} end; cloudi_service_handle_info(Request, State, _Dispatcher) -> ?LOG_WARN("Unknown info \"~p\"", [Request]), {noreply, State}. cloudi_service_terminate(_Reason, _Timeout, undefined) -> ok; cloudi_service_terminate(_Reason, _Timeout, #state{module = Module, connection = Connection}) -> driver_close(Module, Connection), ok. %%%------------------------------------------------------------------------ %%% Private functions %%%------------------------------------------------------------------------ -spec response_external(common_result(), Input :: any(), Endian :: endian()) -> binary(). % rely on an interface result format response_external({updated, Count}, Input, Endian) -> CountBin = if Endian =:= big -> <>; Endian =:= little -> <>; Endian =:= native -> <> end, cloudi_response:new(Input, CountBin, Endian); response_external({selected, Rows}, Input, Endian) -> Count = erlang:length(Rows), CountBin = if Endian =:= big -> <>; Endian =:= little -> <>; Endian =:= native -> <> end, Output = erlang:iolist_to_binary([CountBin | lists:map(fun(T) -> Row = cloudi_string:term_to_list(erlang:tuple_to_list(T)), cloudi_response:new(Input, Row, Endian) end, Rows)]), cloudi_response:new(Input, Output); response_external({error, _} = Error, Input, Endian) -> cloudi_response:new(Input, Error, Endian). %% interface adapter driver_open(?MODULE_EONBLAST, HostName, UserName, Password, Port, Database, TimeoutConnect, Encoding, StartCommands) -> Pool = #pool{pool_id = ?MODULE, size = 1, host = HostName, user = UserName, password = Password, port = Port, database = Database, encoding = Encoding, start_cmds = StartCommands, connect_timeout = TimeoutConnect}, try ?MODULE_EONBLAST:open_connection(Pool) of Connection -> % do not bother to add to Pool#pool.available % (not using the ?MODULE_EONBLAST connection manager process) {ok, Connection} catch ErrorType:Error -> ?LOG_ERROR("start failed ~p ~p~n~p", [ErrorType, Error, erlang:get_stacktrace()]), {error, Error} end. driver_close(?MODULE_EONBLAST, Connection) -> ok = ?MODULE_EONBLAST:close_connection(Connection). % only internal usage, due to relying on an erlang list driver_with_transaction(L, Timeout, State) -> case driver_squery(internal, [<<"BEGIN;">> | L] ++ [<<"COMMIT;">>], Timeout, State#state{interface = common}) of {updated, 0} -> ok; {error, _} = Error -> driver_squery(internal, <<"ROLLBACK;">>, Timeout, State), Error end. % only internal usage, due to relying on an erlang list driver_equery(Query, Parameters, Timeout, #state{module = ?MODULE_EONBLAST, connection = Connection, interface = Interface, debug_level = DebugLevel}) -> NewQuery = if is_atom(Query) -> Query; is_list(Query) -> erlang:iolist_to_binary(Query); is_binary(Query) -> Query end, Native = ?MODULE_EONBLAST:execute(Connection, NewQuery, Parameters, Timeout), if DebugLevel =:= off -> ok; true -> driver_debug(DebugLevel, Query, Parameters, Native) end, if Interface =:= common -> eonblast_to_common(Native); Interface =:= native -> Native end. % internal or external driver_squery(internal, Query, Timeout, #state{module = ?MODULE_EONBLAST, connection = Connection, interface = Interface, debug_level = DebugLevel}) -> QueryBinary = if is_list(Query) -> erlang:iolist_to_binary(Query); is_binary(Query) -> Query end, Native = ?MODULE_EONBLAST:execute(Connection, QueryBinary, [], Timeout), if DebugLevel =:= off -> ok; true -> driver_debug(DebugLevel, Query, Native) end, if Interface =:= common -> eonblast_to_common(Native); Interface =:= native -> Native end; driver_squery(external, Query, Timeout, #state{module = ?MODULE_EONBLAST, connection = Connection, endian = Endian, debug_level = DebugLevel}) -> QueryBinary = if is_list(Query) -> erlang:iolist_to_binary(Query); is_binary(Query) -> Query end, Native = ?MODULE_EONBLAST:execute(Connection, QueryBinary, [], Timeout), if DebugLevel =:= off -> ok; true -> driver_debug(DebugLevel, Query, Native) end, response_external(eonblast_to_common(Native), Query, Endian). driver_prepare(Identifier, Query, Timeout, #state{module = ?MODULE_EONBLAST, connection = Connection, debug_level = DebugLevel}) -> QueryBinary = if is_list(Query) -> erlang:iolist_to_binary(Query); is_binary(Query) -> Query end, Result = try ?MODULE_EONBLAST:preprepare(Connection, Identifier, QueryBinary, Timeout) of ok -> ok catch _:Reason -> {error, Reason} end, if DebugLevel =:= off -> ok; true -> driver_debug_log(DebugLevel, "prepare:~n" " ~p~n" " ~p~n" " = ~p", [Identifier, Query, Result]) end, Result. driver_ping(Timeout, #state{module = ?MODULE_EONBLAST, connection = Connection, debug_level = DebugLevel}) -> Result = ?MODULE_EONBLAST:ping_connection(Connection, Timeout), if DebugLevel =:= off -> ok; true -> driver_debug_log(DebugLevel, "ping:~n" " = ~p", [Result]) end, Result. driver_debug_log(trace, Message, Args) -> ?LOG_TRACE(Message, Args); driver_debug_log(debug, Message, Args) -> ?LOG_DEBUG(Message, Args); driver_debug_log(info, Message, Args) -> ?LOG_INFO(Message, Args); driver_debug_log(warn, Message, Args) -> ?LOG_WARN(Message, Args); driver_debug_log(error, Message, Args) -> ?LOG_ERROR(Message, Args); driver_debug_log(fatal, Message, Args) -> ?LOG_FATAL(Message, Args). driver_debug(Level, Query, Parameters, Result) -> driver_debug_log(Level, "SQL(equery):~n" " ~p~n" " ~p~n" " = ~p", [Query, Parameters, Result]). driver_debug(Level, Query, Result) -> driver_debug_log(Level, "SQL(squery):~n" " ~p~n" " = ~p", [Query, Result]). eonblast_to_common(L) when is_list(L) -> [H | _] = lists:reverse(L), eonblast_to_common(H); eonblast_to_common(#ok_packet{affected_rows = I}) -> {updated, I}; eonblast_to_common(#result_packet{rows = Rows}) -> NewRows = [erlang:list_to_tuple(L) || L <- Rows], {selected, NewRows}; eonblast_to_common(#eof_packet{status = Status}) -> {error, erlang:list_to_binary(?MODULE_EONBLAST:hstate(Status))}; eonblast_to_common(#error_packet{status = Status, msg = Message}) -> Reason = if is_integer(Status) -> erlang:iolist_to_binary([erlang:integer_to_list(Status), ": ", Message]); is_binary(Status) -> erlang:iolist_to_binary([Status, ": ", Message]) end, {error, Reason}.