%%%------------------------------------------------------------------- %%% File : p1_mysql_conn.erl %%% Author : Fredrik Thulin %%% Descrip.: MySQL connection handler, handles de-framing of messages %%% received by the MySQL receiver process. %%% Created : 5 Aug 2005 by Fredrik Thulin %%% Modified: 11 Jan 2006 by Mickael Remond %%% %%% Note : All MySQL code was written by Magnus Ahltorp, originally %%% in the file p1_mysql.erl - I just moved it here. %%% %%% Copyright (c) 2001-2004 Kungliga Tekniska Högskolan %%% See the file COPYING %%% %%% %%% This module handles a single connection to a single MySQL server. %%% You can use it stand-alone, or through the 'p1_mysql' module if you %%% want to have more than one connection to the server, or %%% connections to different servers. %%% %%% To use it stand-alone, set up the connection with %%% %%% {ok, Pid} = p1_mysql_conn:start(Host, Port, User, Password, %%% Database, LogFun) %%% %%% Host = string() %%% Port = integer() %%% User = string() %%% Password = string() %%% Database = string() %%% LogFun = undefined | (gives logging to console) %%% function() of arity 3 (Level, Fmt, Args) %%% %%% Note: In stand-alone mode you have to start Erlang crypto application by %%% yourself with crypto:start() %%% %%% and then make MySQL querys with %%% %%% Result = p1_mysql_conn:fetch(Pid, Query, self()) %%% %%% Result = {data, MySQLRes} | %%% {updated, MySQLRes} | %%% {error, MySQLRes} %%% Where: MySQLRes = #p1_mysql_result %%% %%% Actual data can be extracted from MySQLRes by calling the following API %%% functions: %%% - on data received: %%% FieldInfo = p1_mysql:get_result_field_info(MysqlRes) %%% AllRows = p1_mysql:get_result_rows(MysqlRes) %%% with FieldInfo = list() of {Table, Field, Length, Name} %%% and AllRows = list() of list() representing records %%% - on update: %%% Affected= p1_mysql:get_result_affected_rows(MysqlRes) %%% with Affected = integer() %%% - on error: %%% Reason = p1_mysql:get_result_reason(MysqlRes) %%% with Reason = string() %%%------------------------------------------------------------------- -module(p1_mysql_conn). -behavior(gen_server). -define(CONNECT_TIMEOUT, 5000). %%-------------------------------------------------------------------- %% External exports %%-------------------------------------------------------------------- -export([start/6, start/7, start/8, start_link/6, start_link/7, start_link/8, fetch/3, fetch/4, squery/4, stop/1, init/1, handle_call/3, handle_cast/2, terminate/2, code_change/3, handle_info/2]). %%-------------------------------------------------------------------- %% External exports (should only be used by the 'p1_mysql_auth' module) %%-------------------------------------------------------------------- -export([do_recv/3 ]). -include("p1_mysql.hrl"). -include("p1_mysql_consts.hrl"). -include("p1_mysql_state.hrl"). -include_lib("kernel/include/inet.hrl"). -define(SECURE_CONNECTION, 32768). -define(MYSQL_QUERY_OP, 3). -define(DEFAULT_STANDALONE_TIMEOUT, 5000). -define(DEFAULT_RESULT_TYPE, list). -define(MYSQL_4_0, 40). %% Support for MySQL 4.0.x -define(MYSQL_4_1, 41). %% Support for MySQL 4.1.x et 5.0.x %%==================================================================== %% External functions %%==================================================================== %%-------------------------------------------------------------------- %% Function: start(Host, Port, User, Password, Database, LogFun) %% Function: start_link(Host, Port, User, Password, Database, LogFun) %% Host = string() %% Port = integer() %% User = string() %% Password = string() %% Database = string() %% LogFun = undefined | function() of arity 3 %% Descrip.: Starts a p1_mysql_conn process that connects to a MySQL %% server, logs in and chooses a database. %% Returns : {ok, Pid} | {error, Reason} %% Pid = pid() %% Reason = string() %%-------------------------------------------------------------------- start(Host, Port, User, Password, Database, LogFun) -> start(Host, Port, User, Password, Database, ?CONNECT_TIMEOUT, LogFun). start(Host, Port, User, Password, Database, ConnectTimeout, LogFun) -> start(Host, Port, User, Password, Database, ConnectTimeout, LogFun, []). start(Host, Port, User, Password, Database, ConnectTimeout, LogFun, SSLOpts) when is_list(Host), is_integer(Port), is_list(User), is_list(Password), is_list(Database) -> gen_server:start(?MODULE, [Host, Port, User, Password, Database, ConnectTimeout, LogFun, SSLOpts], [{timeout, ConnectTimeout}]). start_link(Host, Port, User, Password, Database, LogFun) -> start_link(Host, Port, User, Password, Database, ?CONNECT_TIMEOUT, LogFun). start_link(Host, Port, User, Password, Database, ConnectTimeout, LogFun) -> start_link(Host, Port, User, Password, Database, ConnectTimeout, LogFun, []). start_link(Host, Port, User, Password, Database, ConnectTimeout, LogFun, SSLOpts) when is_list(Host), is_integer(Port), is_list(User), is_list(Password), is_list(Database) -> gen_server:start_link(?MODULE, [Host, Port, User, Password, Database, ConnectTimeout, LogFun, SSLOpts], [{timeout, ConnectTimeout}]). %%-------------------------------------------------------------------- %% Function: fetch(Pid, Query, From) %% fetch(Pid, Query, From, Timeout) %% Pid = pid(), p1_mysql_conn to send fetch-request to %% Query = string(), MySQL query in verbatim %% From = pid() or term(), use a From of self() when %% using this module for a single connection, %% or pass the gen_server:call/3 From argument if %% using a gen_server to do the querys (e.g. the %% mysql_dispatcher) %% Timeout = integer() | infinity, gen_server timeout value %% Descrip.: Send a query and wait for the result if running stand- %% alone (From = self()), but don't block the caller if we %% are not running stand-alone (From = gen_server From). %% Returns : ok | (non-stand-alone mode) %% {data, #p1_mysql_result} | (stand-alone mode) %% {updated, #p1_mysql_result} | (stand-alone mode) %% {error, #p1_mysql_result} (stand-alone mode) %% FieldInfo = term() %% Rows = list() of [string()] %% Reason = term() %%-------------------------------------------------------------------- fetch(Pid, Query, From) -> squery(Pid, Query, From, []). fetch(Pid, Query, From, Timeout) -> squery(Pid, Query, From, [{timeout, Timeout}]). squery(Pid, Query, From, Options) when is_pid(Pid), (is_list(Query) or is_binary(Query)) -> Self = self(), Timeout = get_option(timeout, Options, ?DEFAULT_STANDALONE_TIMEOUT), TRef = erlang:make_ref(), Pid ! {fetch, TRef, Query, From, Options}, case From of Self -> %% We are not using a mysql_dispatcher, await the response wait_fetch_result(TRef, Pid, Timeout); _ -> %% From is gen_server From, Pid will do gen_server:reply() %% when it has an answer ok end. wait_fetch_result(TRef, Pid, Timeout) -> receive {'EXIT', Pid, _Reason} -> {error, #p1_mysql_result{error = "connection closed"}}; {fetch_result, TRef, Pid, Result} -> Result; {fetch_result, _BadRef, Pid, _Result} -> wait_fetch_result(TRef, Pid, Timeout) after Timeout -> stop(Pid), timer:kill_after(?DEFAULT_STANDALONE_TIMEOUT, Pid), {error, #p1_mysql_result{error = "query timed out"}} end. stop(Pid) -> Pid ! close. %%-------------------------------------------------------------------- %% Function: do_recv(LogFun, RecvPid, SeqNum) %% LogFun = undefined | function() with arity 3 %% RecvPid = pid(), p1_mysql_recv process %% SeqNum = undefined | integer() %% Descrip.: Wait for a frame decoded and sent to us by RecvPid. %% Either wait for a specific frame if SeqNum is an integer, %% or just any frame if SeqNum is undefined. %% Returns : {ok, Packet, Num} | %% {error, Reason} %% Reason = term() %% %% Note : Only to be used externally by the 'p1_mysql_auth' module. %%-------------------------------------------------------------------- do_recv(LogFun, #state{data = Last} = State, SeqNum) when Last /= <<>> -> case extract_packet(Last) of {Packet, Num, Rest} -> {ok, Packet, Num, State#state{data = Rest}}; _ -> do_recv2(LogFun, State, SeqNum) end; do_recv(LogFun, State, SeqNum) -> do_recv2(LogFun, State, SeqNum). do_recv2(LogFun, #state{socket = {_SockMod, Socket}, data = Last} = State, SeqNum) when is_function(LogFun); LogFun == undefined -> receive close -> %ejabberd_sql:sql_query(<<"localhost">>, <<"select 1+1;">>). p1_mysql:log(LogFun, error, "p1_mysql_conn:" " received close~n", []), {error, "p1_mysql_recv: socket was closed"}; {T, Socket, Data} when T == tcp; T == ssl -> NewData = <>, case extract_packet(NewData) of {Packet, Num, Rest} -> {ok, Packet, Num, State#state{data = Rest}}; Rest -> do_recv(LogFun, State#state{data = Rest}, SeqNum) end; {T, Socket, Reason} when T == tcp_error; T == ssl_error -> p1_mysql:log(LogFun, error, "p1_mysql_conn: " "Socket ~p closed : ~p", [Socket, Reason]), {error, "p1_mysql_recv: socket was closed"}; {T, Socket} when T == tcp_closed; T == ssl_closed -> p1_mysql:log(LogFun, error, "p1_mysql_conn: " "Socket ~p closed", [Socket]), {error, "p1_mysql_recv: socket was closed"}; Other when not is_tuple(Other) orelse size(Other) /= 3 orelse element(1, Other) /= system -> p1_mysql:log(LogFun, error, "p1_mysql_conn: Unknown message ~p~n", [Other]), {error, "p1_mysql_recv: socket was closed"} end. extract_packet(Data) -> case Data of <> -> if Length =< size(D) -> {Packet, Rest} = split_binary(D, Length), {Packet, Num, Rest}; true -> Data end; _ -> Data end. %%==================================================================== %% Internal functions %%==================================================================== %%-------------------------------------------------------------------- %% Function: init(Host, Port, User, Password, Database, LogFun, %% Parent) %% Host = string() %% Port = integer() %% User = string() %% Password = string() %% Database = string() %% LogFun = undefined | function() of arity 3 %% Parent = pid() of process starting this p1_mysql_conn %% Descrip.: Connect to a MySQL server, log in and chooses a database. %% Report result of this to Parent, and then enter loop() if %% we were successfull. %% Returns : void() | does not return %%-------------------------------------------------------------------- init([Host, Port, User, Password, Database, ConnectTimeout, LogFun, SSLOpts]) -> case connect(Host, Port, LogFun, ConnectTimeout) of {ok, Sock0} -> State = #state{socket = {gen_tcp, Sock0}, log_fun = LogFun, data = <<>>}, case mysql_init(State, User, Password, LogFun, SSLOpts) of {ok, NState} -> case do_query(NState, "use " ++ Database, [{result_type, binary}]) of {error, MySQLRes} -> p1_mysql:log(LogFun, error, "p1_mysql_conn: Failed changing" " to database ~p : ~p", [Database, p1_mysql:get_result_reason(MySQLRes)]), {SockMod, RawSock} = NState#state.socket, SockMod:close(RawSock), {stop, normal}; %% ResultType: data | updated {_ResultType, _MySQLRes, NState2} -> {ok, NState2} end; {error, _Reason} -> {stop, normal} end; E -> p1_mysql:log(LogFun, error, "p1_mysql_conn: " "Failed connecting to ~p:~p : ~p", [Host, Port, E]), {stop, normal} end. handle_call(_Request, _From, #state{log_fun = LogFun} = State) -> p1_mysql:log(LogFun, error, "Unhandled call ~p", [_Request]), {reply, {error, badarg}, State}. handle_cast(_Request, #state{log_fun = LogFun} = State) -> p1_mysql:log(LogFun, error, "Unhandled cast ~p", [_Request]), {noreply, State}. handle_info({fetch, Ref, Query, GenSrvFrom, Options}, State) -> %% GenSrvFrom is either a gen_server:call/3 From term(), %% or a pid if no gen_server was used to make the query {NState, Res} = case do_query(State, Query, Options) of {error, R} -> {State, {error, R}}; {T, R, S} -> {S, {T, R}} end, case is_pid(GenSrvFrom) of true -> %% The query was not sent using gen_server mechanisms GenSrvFrom ! {fetch_result, Ref, self(), Res}; false -> %% the timer is canceled in wait_fetch_result/2, but we wait on that funtion only if the query %% was not sent using the mysql gen_server. So we at least should try to cancel the timer here %% (no warranty, the gen_server can still receive timeout messages) erlang:cancel_timer(Ref), gen_server:reply(GenSrvFrom, Res) end, case Res of {error, #p1_mysql_result{error = "p1_mysql_recv: socket was closed"}} -> p1_mysql:log(State#state.log_fun, error, "p1_mysql_conn: " "Connection closed, exiting.", []), {stop, normal, State}; _ -> {noreply, NState} end; handle_info({T, _Socket, Data}, #state{data = Last} = State) when T == tcp; T == ssl -> NewData = <>, {noreply, State#state{data = NewData}}; handle_info({T, Socket, Reason}, #state{log_fun = LogFun} = State) when T == tcp_error; T == ssl_error -> p1_mysql:log(LogFun, error, "p1_mysql_conn: " "Socket ~p closed : ~p", [Socket, Reason]), {stop, normal, State}; handle_info({T, Socket}, #state{log_fun = LogFun} = State) when T == tcp_closed; T == ssl_closed -> p1_mysql:log(LogFun, error, "p1_mysql_conn: " "Socket ~p closed", [Socket]), {stop, normal, State}; handle_info(close, State) -> p1_mysql:log(State#state.log_fun, info, "p1_mysql_conn: " "Received close signal, exiting.", []), {stop, normal, State}; handle_info(Unknown, State) -> p1_mysql:log(State#state.log_fun, error, "p1_mysql_conn: " "Received unknown signal, exiting : ~p", [Unknown]), {stop, normal, State}. terminate(_Reason, State) -> {SockMod, Socket} = State#state.socket, Result = SockMod:close(Socket), p1_mysql:log(State#state.log_fun, normal, "Closing connection ~p: ~p~n", [State#state.socket, Result]), ok. code_change(_OldVsn, State, _Extra) -> {ok, State}. %%-------------------------------------------------------------------- %% Function: mysql_init(Sock, RecvPid, User, Password, LogFun) %% Sock = term(), gen_tcp socket %% RecvPid = pid(), p1_mysql_recv process %% User = string() %% Password = string() %% LogFun = undefined | function() with arity 3 %% Descrip.: Try to authenticate on our new socket. %% Returns : {ok, SockPair, Version} | {error, Reason} %% Reason = string() %%-------------------------------------------------------------------- mysql_init(State, User, Password, LogFun, SSLOpts) -> case do_recv(LogFun, State, undefined) of {ok, Packet, InitSeqNum, NState} -> case greeting(Packet, LogFun) of {ok, {Version, Salt, Caps, AuthPlug}} -> case Caps band ?CLIENT_SSL of 0 -> case proplists:get_bool(ssl_required, SSLOpts) of true -> p1_mysql:log(LogFun, error, "p1_mysql_conn: " "init failed - ssl required, but not available~n", []), {error, "SSL not available"}; false -> authenticate(NState, User, Password, LogFun, InitSeqNum, Version, Salt, Caps, AuthPlug) end; _ -> case proplists:get_bool(ssl, SSLOpts) orelse proplists:get_bool(ssl_required, SSLOpts) of true -> case start_ssl(NState, SSLOpts, LogFun, InitSeqNum + 1, AuthPlug) of {ok, NewState} -> authenticate(NewState, User, Password, LogFun, InitSeqNum + 1, Version, Salt, Caps, AuthPlug); {error, Reason} -> {error, Reason} end; _ -> authenticate(NState, User, Password, LogFun, InitSeqNum, Version, Salt, Caps, AuthPlug) end end; {error, Reason} -> {error, Reason} end; {error, Reason} -> {error, Reason} end. %% part of mysql_init/4 start_ssl(#state{socket = {_, Sock}} = State, SSLOpts, LogFun, SeqNum, AuthPlug) -> Packet = p1_mysql_auth:get_auth_head(AuthPlug, ?CLIENT_SSL), Data = <<(size(Packet)):24/little, SeqNum:8, Packet/binary>>, p1_mysql:log(LogFun, debug, "p1_mysql_conn send start ssl ~p: ~p", [SeqNum, Packet]), gen_tcp:send(Sock, Data), {[A, B, C], _} = proplists:split(SSLOpts, [certfile, cacertfile, verify]), Filtered = A ++ B ++ C, case ssl:connect(Sock, [binary, {packet, 0} | Filtered]) of {ok, SSLSock} -> {ok, State#state{socket = {ssl, SSLSock}}}; {error, Reason} -> p1_mysql:log(LogFun, error, "p1_mysql_conn: " "ssl start failed: ~p~n", [Reason]), {error, "ssl failed"} end. authenticate(State, User, Password, LogFun, SeqNum, Version, Salt, Caps, AuthPlug) -> AuthRes = p1_mysql_auth:do_auth(AuthPlug, State, SeqNum + 1, User, Password, Salt, Caps, LogFun), case AuthRes of {ok, <<0:8, _Rest/binary>>, _RecvNum, NState} -> {ok, NState#state{mysql_version = Version}}; {ok, <<255:8, Code:16/little, Message/binary>>, _RecvNum, _NState} -> p1_mysql:log(LogFun, error, "p1_mysql_conn: " "init error ~p: ~p~n", [Code, binary_to_list(Message)]), {error, binary_to_list(Message)}; {ok, RecvPacket, _RecvNum, _NState} -> p1_mysql:log(LogFun, error, "p1_mysql_conn: " "init unknown error ~p~n", [binary_to_list(RecvPacket)]), {error, binary_to_list(RecvPacket)}; {error, Reason} -> p1_mysql:log(LogFun, error, "p1_mysql_conn: " "init failed receiving data : ~p~n", [Reason]), {error, Reason} end. greeting(Packet, LogFun) -> <> = Packet, case Protocol of 9 -> {ServerStatusStr, Rest2} = asciz(Rest), <<_TreadID:32/little, Rest3/binary>> = Rest2, {Salt, _} = asciz(Rest3), p1_mysql:log(LogFun, debug, "p1_mysql_conn: greeting version ~p (protocol ~p) " "salt ~p", [ServerStatusStr, Protocol, Salt]), {ok, {?MYSQL_4_0, Salt, 0, "old_pass"}}; 10 -> {ServerStatusStr, Rest2} = asciz(Rest), <<_TreadID:32/little, Salt1:8/binary, _:8, Caps1:16/little, CharSet:8, _StatusFlags:16/little, Caps2:16/little, AuthPlugLen:8, _:10/binary, Rest3/binary>> = Rest2, Caps = (Caps2 bsl 16) bor Caps1, {Salt2, AuthPlug} = case {Caps band ?CLIENT_PLUGIN_AUTH, AuthPlugLen} of {0, 0} -> Len = max(13, AuthPlugLen - 8) - 1, <> = Rest3, {S, "mysql_native_password"}; {?CLIENT_PLUGIN_AUTH, _} -> Len = max(13, AuthPlugLen - 8) - 1, <> = Rest3, {AuthPlugName, _} = asciz(Rest4), {S, AuthPlugName} end, Salt = binary_to_list(<>), p1_mysql:log(LogFun, debug, "p1_mysql_conn: greeting version ~p (protocol ~p) " "salt ~p caps ~p serverchar ~p auth_plug: ~p", [ServerStatusStr, Protocol, Salt, Caps, CharSet, AuthPlug]), {ok, {?MYSQL_4_1, Salt, Caps, AuthPlug}}; 255 -> case Rest of <> -> {error, io_lib:format("p1_mysql_conn: greetings error code ~p", [Code])}; <<_Code:16/little, Msg/binary>> -> {error, io_lib:format("p1_mysql_conn: greetings error: ~s", [Msg])}; _ -> {error, "p1_mysql_conn: greetings error"} end end. %% part of greeting/2 asciz(Data) -> p1_mysql:asciz_binary(Data, []). %%-------------------------------------------------------------------- %% Function: get_query_response(LogFun, RecvPid) %% LogFun = undefined | function() with arity 3 %% RecvPid = pid(), p1_mysql_recv process %% Version = integer(), Representing MySQL version used %% Descrip.: Wait for frames until we have a complete query response. %% Returns : {data, #p1_mysql_result} %% {updated, #p1_mysql_result} %% {error, #p1_mysql_result} %% FieldInfo = list() of term() %% Rows = list() of [string()] %% AffectedRows = int() %% Reason = term() %%-------------------------------------------------------------------- get_query_response(LogFun, State, Version, Options) -> case do_recv(LogFun, State, undefined) of {ok, <>, _, NState} -> case Fieldcount of 0 -> %% No Tabular data AffectedRows = case Rest of <<16#fc, Value:16/little, _/binary>> -> Value; <<16#fd, Value:24/little, _/binary>> -> Value; <<16#fe, Value:64/little, _/binary>> -> Value; <> -> Value end, {updated, #p1_mysql_result{affectedrows = AffectedRows}, NState}; 255 -> <<_Code:16/little, Message/binary>> = Rest, {error, #p1_mysql_result{error = binary_to_list(Message)}}; _ -> %% Tabular data received ResultType = get_option(result_type, Options, ?DEFAULT_RESULT_TYPE), case get_fields(LogFun, NState, [], Version, ResultType) of {ok, Fields, NState2} -> case get_rows(Fieldcount, LogFun, NState2, ResultType, []) of {ok, Rows, NState3} -> {data, #p1_mysql_result{fieldinfo = Fields, rows = Rows}, NState3}; {error, Reason} -> {error, #p1_mysql_result{error = Reason}} end; {error, Reason} -> {error, #p1_mysql_result{error = Reason}} end end; {error, Reason} -> {error, #p1_mysql_result{error = Reason}} end. %%-------------------------------------------------------------------- %% Function: get_fields(LogFun, RecvPid, [], Version) %% LogFun = undefined | function() with arity 3 %% RecvPid = pid(), p1_mysql_recv process %% Version = integer(), Representing MySQL version used %% Descrip.: Received and decode field information. %% Returns : {ok, FieldInfo} | %% {error, Reason} %% FieldInfo = list() of term() %% Reason = term() %%-------------------------------------------------------------------- %% Support for MySQL 4.0.x: get_fields(LogFun, State, Res, ?MYSQL_4_0, ResultType) -> case do_recv(LogFun, State, undefined) of {ok, Packet, _Num, NState} -> case Packet of <<254:8>> -> {ok, lists:reverse(Res), NState}; <<254:8, Rest/binary>> when size(Rest) < 8 -> {ok, lists:reverse(Res), NState}; _ -> {Table, Rest} = get_with_length(Packet), {Field, Rest2} = get_with_length(Rest), {LengthB, Rest3} = get_with_length(Rest2), LengthL = size(LengthB)*8, <> = LengthB, {Type, Rest4} = get_with_length(Rest3), {_Flags, _Rest5} = get_with_length(Rest4), if ResultType == list -> This = {binary_to_list(Table), binary_to_list(Field), Length, %% TODO: Check on MySQL 4.0 if types are specified %% using the same 4.1 formalism and could %% be expanded to atoms: binary_to_list(Type)}; ResultType == binary -> This = {Table, Field, Length, Type} end, get_fields(LogFun, NState, [This | Res], ?MYSQL_4_0, ResultType) end; {error, Reason} -> {error, Reason} end; %% Support for MySQL 4.1.x and 5.x: get_fields(LogFun, State, Res, ?MYSQL_4_1, ResultType) -> case do_recv(LogFun, State, undefined) of {ok, Packet, _Num, NState} -> case Packet of <<254:8>> -> {ok, lists:reverse(Res), NState}; <<254:8, Rest/binary>> when size(Rest) < 8 -> {ok, lists:reverse(Res), NState}; _ -> {_Catalog, Rest} = get_with_length(Packet), {_Database, Rest2} = get_with_length(Rest), {Table, Rest3} = get_with_length(Rest2), %% OrgTable is the real table name if Table is an alias {_OrgTable, Rest4} = get_with_length(Rest3), {Field, Rest5} = get_with_length(Rest4), %% OrgField is the real field name if Field is an alias {_OrgField, Rest6} = get_with_length(Rest5), <<_Metadata:8/little, _Charset:16/little, Length:32/little, Type:8/little, _Flags:16/little, _Decimals:8/little, _Rest7/binary>> = Rest6, if ResultType == list -> This = {binary_to_list(Table), binary_to_list(Field), Length, get_field_datatype(Type)}; ResultType == binary -> This = {Table, Field, Length, get_field_datatype(Type)} end, get_fields(LogFun, NState, [This | Res], ?MYSQL_4_1, ResultType) end; {error, Reason} -> {error, Reason} end. %%-------------------------------------------------------------------- %% Function: get_rows(N, LogFun, RecvPid, []) %% N = integer(), number of rows to get %% LogFun = undefined | function() with arity 3 %% RecvPid = pid(), p1_mysql_recv process %% Descrip.: Receive and decode a number of rows. %% Returns : {ok, Rows} | %% {error, Reason} %% Rows = list() of [string()] %%-------------------------------------------------------------------- get_rows(N, LogFun, State, ResultType, Res) -> case do_recv(LogFun, State, undefined) of {ok, Packet, _Num, NState} -> case Packet of <<254:8, Rest/binary>> when size(Rest) < 8 -> {ok, lists:reverse(Res), NState}; _ -> {ok, This} = get_row(N, Packet, ResultType, []), get_rows(N, LogFun, NState, ResultType, [This | Res]) end; {error, Reason} -> {error, Reason} end. %% part of get_rows/4 get_row(0, _Data, _ResultType, Res) -> {ok, lists:reverse(Res)}; get_row(N, Data, ResultType, Res) -> {Col, Rest} = get_with_length(Data), This = case Col of null -> null; _ -> if ResultType == list -> binary_to_list(Col); ResultType == binary -> Col end end, get_row(N - 1, Rest, ResultType, [This | Res]). get_with_length(<<251:8, Rest/binary>>) -> {null, Rest}; get_with_length(<<252:8, Length:16/little, Rest/binary>>) -> split_binary(Rest, Length); get_with_length(<<253:8, Length:24/little, Rest/binary>>) -> split_binary(Rest, Length); get_with_length(<<254:8, Length:64/little, Rest/binary>>) -> split_binary(Rest, Length); get_with_length(<>) when Length < 251 -> split_binary(Rest, Length). %%-------------------------------------------------------------------- %% Function: do_query(State, Query) %% do_query(Sock, RecvPid, LogFun, Query) %% Sock = term(), gen_tcp socket %% RecvPid = pid(), p1_mysql_recv process %% LogFun = undefined | function() with arity 3 %% Query = string() %% Descrip.: Send a MySQL query and block awaiting it's response. %% Returns : result of get_query_response/2 | {error, Reason} %%-------------------------------------------------------------------- do_query(#state{socket = Sock, log_fun = LogFun, mysql_version = Version} = State, Query, Options) when (is_list(Query) or is_binary(Query)) -> Packet = list_to_binary([?MYSQL_QUERY_OP, Query]), case do_send(Sock, Packet, 0, LogFun) of ok -> get_query_response(LogFun, State, Version, Options); {error, Reason} -> Msg = io_lib:format("Failed sending data on socket : ~p", [Reason]), {error, Msg} end. %%-------------------------------------------------------------------- %% Function: do_send(Sock, Packet, SeqNum, LogFun) %% Sock = {SockMod, Socket}, socket %% Packet = binary() %% SeqNum = integer(), packet sequence number %% LogFun = undefined | function() with arity 3 %% Descrip.: Send a packet to the MySQL server. %% Returns : result of gen_tcp:send/2 %%-------------------------------------------------------------------- do_send({SockMod, Sock}, Packet, SeqNum, _LogFun) when is_binary(Packet), is_integer(SeqNum) -> Data = <<(size(Packet)):24/little, SeqNum:8, Packet/binary>>, %p1_mysql:log(_LogFun, debug, "p1_mysql_conn: send packet ~p: ~p", %[SeqNum, Data]), SockMod:send(Sock, Data). %%-------------------------------------------------------------------- %% Function: get_field_datatype(DataType) %% DataType = integer(), MySQL datatype %% Descrip.: Return MySQL field datatype as description string %% Returns : String, MySQL datatype %%-------------------------------------------------------------------- get_field_datatype(0) -> 'DECIMAL'; get_field_datatype(1) -> 'TINY'; get_field_datatype(2) -> 'SHORT'; get_field_datatype(3) -> 'LONG'; get_field_datatype(4) -> 'FLOAT'; get_field_datatype(5) -> 'DOUBLE'; get_field_datatype(6) -> 'NULL'; get_field_datatype(7) -> 'TIMESTAMP'; get_field_datatype(8) -> 'LONGLONG'; get_field_datatype(9) -> 'INT24'; get_field_datatype(10) -> 'DATE'; get_field_datatype(11) -> 'TIME'; get_field_datatype(12) -> 'DATETIME'; get_field_datatype(13) -> 'YEAR'; get_field_datatype(14) -> 'NEWDATE'; get_field_datatype(16) -> 'BIT'; get_field_datatype(246) -> 'DECIMAL'; get_field_datatype(247) -> 'ENUM'; get_field_datatype(248) -> 'SET'; get_field_datatype(249) -> 'TINYBLOB'; get_field_datatype(250) -> 'MEDIUM_BLOG'; get_field_datatype(251) -> 'LONG_BLOG'; get_field_datatype(252) -> 'BLOB'; get_field_datatype(253) -> 'VAR_STRING'; get_field_datatype(254) -> 'STRING'; get_field_datatype(255) -> 'GEOMETRY'. %%-------------------------------------------------------------------- %% Function: get_option(Key1, Options, Default) -> Value1 %% Options = [Option] %% Option = {Key2, Value2} %% Key1 = Key2 = atom() %% Value1 = Value2 = Default = term() %% Descrip.: Return the option associated with Key passed to squery/4 %%-------------------------------------------------------------------- get_option(Key, Options, Default) -> case lists:keysearch(Key, 1, Options) of {value, {_, Value}} -> Value; false -> Default end. %%-------------------------------------------------------------------- %% Connecting stuff %%-------------------------------------------------------------------- connect(Host, Port, LogFun, Timeout) -> case lookup(Host, Timeout) of {ok, AddrsFamilies} -> do_connect(AddrsFamilies, Port, {error, nxdomain}, Timeout); {error, E} -> Reason = inet:format_error(E), p1_mysql:log(LogFun, error, "p1_mysql_conn: Failed connecting to ~s:~p: ~s", [Host, Port, Reason]), Msg = lists:flatten(io_lib:format("connect failed: ~s", [Reason])), {error, Msg} end. do_connect([{IP, Family} | AddrsFamilies], Port, _Err, Timeout) -> case gen_tcp:connect(IP, Port, [binary, {packet, 0}, {active, true}, Family], Timeout) of {ok, Sock} -> {ok, Sock}; {error, _} = Err -> do_connect(AddrsFamilies, Port, Err, Timeout) end; do_connect([], _Port, Err, _Timeout) -> Err. lookup(Host, Timeout) -> case inet:parse_address(Host) of {ok, IP} -> {ok, [{IP, get_addr_type(IP)}]}; {error, _} -> do_lookup([{Host, Family} || Family <- [inet6, inet]], [], {error, nxdomain}, Timeout) end. do_lookup([{Host, Family} | HostFamilies], AddrFamilies, Err, Timeout) -> case inet:gethostbyname(Host, Family, Timeout) of {ok, HostEntry} -> Addrs = host_entry_to_addrs(HostEntry), AddrFamilies1 = [{Addr, Family} || Addr <- Addrs], do_lookup(HostFamilies, AddrFamilies ++ AddrFamilies1, Err, Timeout); {error, _} = Err1 -> do_lookup(HostFamilies, AddrFamilies, Err1, Timeout) end; do_lookup([], [], Err, _Timeout) -> Err; do_lookup([], AddrFamilies, _Err, _Timeout) -> {ok, AddrFamilies}. host_entry_to_addrs(#hostent{h_addr_list = AddrList}) -> lists:filter( fun(Addr) -> try get_addr_type(Addr) of _ -> true catch _:badarg -> false end end, AddrList). get_addr_type({_, _, _, _}) -> inet; get_addr_type({_, _, _, _, _, _, _, _}) -> inet6; get_addr_type(_) -> erlang:error(badarg).