diff options
| -rw-r--r-- | src/mysqlerl.c | 9 | ||||
| -rw-r--r-- | src/mysqlerl_connection.erl | 38 | ||||
| -rw-r--r-- | src/mysqlerl_port.erl | 54 | ||||
| -rw-r--r-- | src/mysqlerl_port.hrl | 1 | ||||
| -rw-r--r-- | src/mysqlerl_port_sup.erl | 16 |
5 files changed, 92 insertions, 26 deletions
diff --git a/src/mysqlerl.c b/src/mysqlerl.c index b2a6ab3..5cec054 100644 --- a/src/mysqlerl.c +++ b/src/mysqlerl.c @@ -13,6 +13,7 @@ #include <string.h> const char *QUERY_MSG = "sql_query"; +const char *PARAM_QUERY_MSG = "param_query_msg"; void usage() @@ -173,6 +174,11 @@ handle_sql_query(MYSQL *dbh, ETERM *cmd) } void +handle_sql_param_query(MYSQL *dbh, ETERM *cmd) +{ +} + +void dispatch_db_cmd(MYSQL *dbh, ETERM *msg) { ETERM *tag; @@ -181,6 +187,9 @@ dispatch_db_cmd(MYSQL *dbh, ETERM *msg) if (strncmp((char *)ERL_ATOM_PTR(tag), QUERY_MSG, strlen(QUERY_MSG)) == 0) { handle_sql_query(dbh, msg); + } else if (strncmp((char *)ERL_ATOM_PTR(tag), + PARAM_QUERY_MSG, strlen(PARAM_QUERY_MSG)) == 0) { + handle_sql_param_query(dbh, msg); } else { logmsg("WARNING: message type %s unknown.", (char *)ERL_ATOM_PTR(tag)); erl_free_term(tag); diff --git a/src/mysqlerl_connection.erl b/src/mysqlerl_connection.erl index 495f841..b6015e3 100644 --- a/src/mysqlerl_connection.erl +++ b/src/mysqlerl_connection.erl @@ -2,6 +2,7 @@ -author('bjc@kublai.com'). -include("mysqlerl.hrl"). +-include("mysqlerl_port.hrl"). -behavior(gen_server). @@ -10,8 +11,7 @@ -export([init/1, terminate/2, code_change/3, handle_call/3, handle_cast/2, handle_info/2]). --record(state, {ref, owner}). --record(port_closed, {reason}). +-record(state, {sup, owner}). start_link(Owner, Host, Port, Database, User, Password, Options) -> gen_server:start_link(?MODULE, [Owner, Host, Port, Database, @@ -26,16 +26,11 @@ init([Owner, Host, Port, Database, User, Password, Options]) -> Cmd = lists:flatten(io_lib:format("~s ~s ~w ~s ~s ~s ~s", [helper(), Host, Port, Database, User, Password, Options])), - Ref = open_port({spawn, Cmd}, [{packet, 4}, binary]), - {ok, #state{ref = Ref, owner = Owner}}. + {ok, Sup} = mysqlerl_port_sup:start_link(Cmd), + {ok, #state{sup = Sup, owner = Owner}}. -terminate(#port_closed{reason = Reason}, #state{ref = Ref}) -> - io:format("DEBUG: mysqlerl connection ~p shutting down (~p).~n", - [Ref, Reason]), - ok; -terminate(Reason, State) -> - port_close(State#state.ref), - io:format("DEBUG: got terminate: ~p~n", [Reason]), +terminate(Reason, _State) -> + io:format("DEBUG: connection got terminate: ~p~n", [Reason]), ok. code_change(_OldVsn, State, _Extra) -> @@ -49,17 +44,15 @@ handle_call(Request, From, #state{owner = Owner} = State) handle_call(stop, _From, State) -> {stop, normal, State}; handle_call(Request, _From, State) -> - {reply, make_request(State#state.ref, Request), State}. + {reply, gen_server:call(port_ref(State#state.sup), + #req{request = Request}), State}. handle_cast(_Request, State) -> {noreply, State}. handle_info({'EXIT', Pid, _Reason}, #state{owner = Pid} = State) -> io:format("DEBUG: owner ~p shut down.~n", [Pid]), - {stop, normal, State}; -handle_info({'EXIT', Ref, Reason}, #state{ref = Ref} = State) -> - io:format("DEBUG: Port ~p closed on ~p.~n", [Ref, State]), - {stop, #port_closed{reason = Reason}, State}. + {stop, normal, State}. helper() -> case code:priv_dir(mysqlerl) of @@ -68,13 +61,6 @@ helper() -> end, filename:join([PrivDir, "mysqlerl"]). -make_request(Ref, Req) -> - io:format("DEBUG: Sending request: ~p~n", [Req]), - port_command(Ref, term_to_binary(Req)), - receive - {Ref, {data, Res}} -> binary_to_term(Res); - Other -> - error_logger:warning_msg("Got unknown query response: ~p~n", - [Other]), - exit({badreply, Other}) - end. +port_ref(Sup) -> + [{mysqlerl_port, Ref, worker, _}] = supervisor:which_children(Sup), + Ref. diff --git a/src/mysqlerl_port.erl b/src/mysqlerl_port.erl new file mode 100644 index 0000000..afcdbf8 --- /dev/null +++ b/src/mysqlerl_port.erl @@ -0,0 +1,54 @@ +-module(mysqlerl_port). +-author('bjc@kublai.com'). + +-include("mysqlerl_port.hrl"). + +-behavior(gen_server). + +-export([start_link/1]). +-export([init/1, terminate/2, code_change/3, + handle_call/3, handle_cast/2, handle_info/2]). + +-record(state, {ref}). +-record(port_closed, {reason}). + +start_link(Cmd) -> + gen_server:start_link(?MODULE, [Cmd], []). + +init([Cmd]) -> + process_flag(trap_exit, true), + Ref = open_port({spawn, Cmd}, [{packet, 4}, binary]), + {ok, #state{ref = Ref}}. + +terminate(#port_closed{reason = Reason}, #state{ref = Ref}) -> + io:format("DEBUG: mysqlerl connection ~p shutting down (~p).~n", + [Ref, Reason]), + ok; +terminate(Reason, State) -> + catch port_close(State#state.ref), + io:format("DEBUG: mysqlerl_port got terminate: ~p~n", [Reason]), + ok. + +code_change(_OldVsn, State, _Extra) -> + {ok, State}. + +handle_call(#req{request = Request}, _From, State) -> + {reply, make_request(State#state.ref, Request), State}. + +handle_cast(_Request, State) -> + {noreply, State}. + +handle_info({'EXIT', Ref, Reason}, #state{ref = Ref} = State) -> + io:format("DEBUG: Port ~p closed on ~p.~n", [Ref, State]), + {stop, #port_closed{reason = Reason}, State}. + +make_request(Ref, Req) -> + io:format("DEBUG: Sending request: ~p~n", [Req]), + port_command(Ref, term_to_binary(Req)), + receive + {Ref, {data, Res}} -> binary_to_term(Res); + Other -> + error_logger:warning_msg("Got unknown query response: ~p~n", + [Other]), + exit({badreply, Other}) + end. diff --git a/src/mysqlerl_port.hrl b/src/mysqlerl_port.hrl new file mode 100644 index 0000000..a7a3300 --- /dev/null +++ b/src/mysqlerl_port.hrl @@ -0,0 +1 @@ +-record(req, {request}). diff --git a/src/mysqlerl_port_sup.erl b/src/mysqlerl_port_sup.erl new file mode 100644 index 0000000..820f929 --- /dev/null +++ b/src/mysqlerl_port_sup.erl @@ -0,0 +1,16 @@ +-module(mysqlerl_port_sup). +-author('bjc@kublai.com'). + +-behavior(supervisor). + +-export([start_link/1]). +-export([init/1]). + +start_link(Cmd) -> + supervisor:start_link(?MODULE, [Cmd]). + +init([Cmd]) -> + Port = {mysqlerl_port, {mysqlerl_port, start_link, [Cmd]}, + transient, 5, worker, [mysqlerl_port]}, + {ok, {{one_for_one, 10, 5}, + [Port]}}. |