%%%------------------------------------------------------------------- %%% @author Niclas Axelsson %%% @copyright (C) 2014-2015, Niclas Axelsson %%% @doc %%% MySQL adapter for ErlDB %%% @end %%% Created : 23 Dec 2014 by Niclas Axelsson %%% Updated : 4 Apr 2015 by Niclas Axelsson %%%------------------------------------------------------------------- -module(erldb_mysql). -behaviour(gen_server). %% API -export([start_link/1]). %% gen_server callbacks -export([ init/1, handle_call/3, handle_cast/2, handle_info/2, terminate/2, code_change/3 ]). -export([ create_table/3 ]). -define(SERVER, ?MODULE). -record(state, { connection :: pid() }). %%%=================================================================== %%% API %%%=================================================================== %%-------------------------------------------------------------------- %% @doc %% Starts the server %% %% @spec start_link() -> {ok, Pid} | ignore | {error, Error} %% @end %%-------------------------------------------------------------------- start_link(Args) -> WorkerArgs = proplists:get_value(worker_args, Args, []), gen_server:start_link(?MODULE, WorkerArgs, []). %%%=================================================================== %%% gen_server callbacks %%%=================================================================== %%-------------------------------------------------------------------- %% @private %% @doc %% Initializes the server %% %% @spec init(Args) -> {ok, State} | %% {ok, State, Timeout} | %% ignore | %% {stop, Reason} %% @end %%-------------------------------------------------------------------- init(Args) -> DBHost = proplists:get_value(db_host, Args, "localhost"), DBPort = proplists:get_value(db_port, Args, 3306), DBUsername = proplists:get_value(db_username, Args, "guest"), DBPassword = proplists:get_value(db_password, Args, ""), DBDatabase = proplists:get_value(db_database, Args, "test"), {ok, Conn} = mysql_conn:start_link(DBHost, DBPort, DBUsername, DBPassword, DBDatabase, fun(_, _, _, _) -> ok end, utf8, erldb_pool), {ok, #state{ connection = Conn }}. %%-------------------------------------------------------------------- %% @private %% @doc %% Handling call messages %% %% @spec handle_call(Request, From, State) -> %% {reply, Reply, State} | %% {reply, Reply, State, Timeout} | %% {noreply, State} | %% {noreply, State, Timeout} | %% {stop, Reason, Reply, State} | %% {stop, Reason, State} %% @end %%-------------------------------------------------------------------- handle_call({init_table, Model, _Args}, _From, State) -> Fields = erldb_lib:get_fields(Model), Res = create_table(State#state.connection, Model, Fields), {reply, Res, State}; handle_call({find, Model, Conditions, Options}, _From, State) -> Query = build_select_query(Model, Conditions, Options), Res = case fetch(State#state.connection, Query) of {data, Data} -> MData = mysql:get_result_rows(Data), [ list_to_tuple([Model|[ unpack_value(Y) || Y <- X ]]) || X <- MData ]; Error -> {error, Error} end, {reply, Res, State}; handle_call({Action, Object}, _From, State) when Action == save orelse Action == update -> Model = erlang:element(1, Object), Zipper = erldb_lib:unzip_object(Object), ModelStr = atom_to_binary(Model, utf8), SQL = case Action of save -> {Fields, Values} = lists:foldl( fun({'id', 'id', _Type, _Args}, Acc) -> Acc; ({'id', Val, Type, Args}, {Attrs, Vals}) -> {["id"|Attrs], [pack_value(Val, Type, Args)|Vals]}; ({Field, Val, Type, Args}, {Attrs, Vals}) -> {[atom_to_list(Field)|Attrs], [pack_value(Val, Type, Args)|Vals]} end, {[], []}, Zipper), FieldsStr = list_to_binary(string:join(escape_attr(Fields), ", ")), [ <<"INSERT INTO ">>, ModelStr, <<" (">>, FieldsStr, <<") VALUES (">>, Values, <<")">> ]; update -> SetFields = lists:foldl( fun({'id', _, _, _}, Aux) -> Aux; ({Field, Val, _, _}, Aux) -> [lists:concat(["`", Field, "`=", Val])|Aux] end, [], Zipper), {value, {_, IdVal, _, _}} = lists:keysearch('id', 1, Zipper), [ <<"UPDATE " >>, ModelStr, <<"SET ">>, string:join(SetFields, ", "), <<"WHERE id=">>, IdVal ] end, Res = fetch(State#state.connection, SQL), {reply, {ok, Res}, State}; handle_call({supported_condition, _Conditions}, _From, State) -> {reply, {ok, supported}, State}; handle_call({supported_operation, Operation}, _From, State) -> Reply = case Operation of find_one -> {error, op_not_supported}; _ -> {ok, supported} end, {reply, Reply, State}; handle_call(_Request, _From, State) -> Reply = ok, {reply, Reply, State}. %%-------------------------------------------------------------------- %% @private %% @doc %% Handling cast messages %% %% @spec handle_cast(Msg, State) -> {noreply, State} | %% {noreply, State, Timeout} | %% {stop, Reason, State} %% @end %%-------------------------------------------------------------------- handle_cast(_Msg, State) -> {noreply, State}. %%-------------------------------------------------------------------- %% @private %% @doc %% Handling all non call/cast messages %% %% @spec handle_info(Info, State) -> {noreply, State} | %% {noreply, State, Timeout} | %% {stop, Reason, State} %% @end %%-------------------------------------------------------------------- handle_info(_Info, State) -> {noreply, State}. %%-------------------------------------------------------------------- %% @private %% @doc %% This function is called by a gen_server when it is about to %% terminate. It should be the opposite of Module:init/1 and do any %% necessary cleaning up. When it returns, the gen_server terminates %% with Reason. The return value is ignored. %% %% @spec terminate(Reason, State) -> void() %% @end %%-------------------------------------------------------------------- terminate(_Reason, _State) -> ok. %%-------------------------------------------------------------------- %% @private %% @doc %% Convert process state when code is changed %% %% @spec code_change(OldVsn, State, Extra) -> {ok, NewState} %% @end %%-------------------------------------------------------------------- code_change(_OldVsn, State, _Extra) -> {ok, State}. %%%=================================================================== %%% Internal functions %%%=================================================================== fetch(Conn, Query) -> case mysql_conn:fetch(Conn, [Query], self()) of {error, _MysqlRes} = Err -> Err; Result -> Result end. create_table(Conn, TableName, TableDefinition) when is_atom(TableName) -> SQL = lists:flatten(["CREATE TABLE IF NOT EXISTS ", erlang:atom_to_list(TableName), " ( ", string:join(table_definition_to_sql_columns(TableDefinition), ","), " )"]), fetch(Conn, SQL). build_select_query(Type, Conditions, Options) -> ["SELECT * FROM ", erlang:atom_to_list(Type), case Conditions of [] -> ""; _ -> " WHERE " ++ string:join(build_conditions(Conditions), " AND ") end, " ORDER BY ", atom_to_list(proplists:get_value(order_by, Options, id)), " ", atom_to_list(proplists:get_value(order, Options, desc)), case proplists:get_value(limit, Options) of undefined -> ""; {Page, Max} -> [" LIMIT ", integer_to_list(Page), ", ", integer_to_list(Max)]; Max -> [" LIMIT ", integer_to_list(Max)] end]. table_definition_to_sql_columns([]) -> ""; table_definition_to_sql_columns([{Fieldname, _Pos, Fieldtype, Args}|Tl]) -> Length = case {Fieldtype, proplists:get_value(length, Args)} of {_, Val} when is_integer(Val) -> "(" ++ erlang:integer_to_list(Val) ++ ")"; {string, undefined} -> "(256)"; _ -> "" end, ["`" ++ erlang:atom_to_list(Fieldname) ++ "` " ++ type_to_sql(Fieldtype) ++ Length ++ " " ++ string:join(options_to_sql(Args), " ")|table_definition_to_sql_columns(Tl)]. escape_attr(Attributes) -> [ ["`", Attribute, "`"] || Attribute <- Attributes ]. build_conditions([]) -> []; build_conditions([{Field, 'equals', Value}|Tl]) -> ["`" ++ atom_to_list(Field) ++ "`=\"" ++ Value ++ "\""|build_conditions(Tl)]. options_to_sql([]) -> []; options_to_sql([{length, _}|Tl]) -> options_to_sql(Tl); options_to_sql([{default, Value}|Tl]) when is_list(Value) -> ["DEFAULT \"", Value, "\""|options_to_sql(Tl)]; options_to_sql([{default, Value}|Tl]) -> ["DEFAULT ", pack_value(Value, false, false)|options_to_sql(Tl)]; options_to_sql([not_null|Tl]) -> ["NOT NULL"|options_to_sql(Tl)]; options_to_sql([null|Tl]) -> ["NULL"|options_to_sql(Tl)]; options_to_sql([auto_increment|Tl]) -> ["AUTO_INCREMENT"|options_to_sql(Tl)]; options_to_sql([primary_key|Tl]) -> ["PRIMARY KEY"|options_to_sql(Tl)]; options_to_sql([unique|Tl]) -> ["UNIQUE KEY"|options_to_sql(Tl)]; options_to_sql([{storage, Type}|Tl]) -> Res = case Type of disk -> "DISK"; memory -> "MEMORY"; default -> "DEFAULT" end, ["STORAGE ", Res|options_to_sql(Tl)]. unpack_value(<<"FALSE">>) -> false; unpack_value(<<"TRUE">>) -> true; unpack_value(Int) when is_integer(Int) -> Int; unpack_value(MaybeTerm) when is_binary(MaybeTerm) -> case catch erlang:binary_to_term(MaybeTerm) of {'EXIT', _} -> binary_to_list(MaybeTerm); Res -> Res end; unpack_value(undefined) -> undefined; unpack_value(Atom) when is_atom(Atom) -> atom_to_list(Atom); unpack_value(_) -> throw(error). pack_value(undefined, _Type, Args) -> proplists:get_value(default, Args, "NULL"); pack_value(Term, erlang_term, _Args) -> term_to_binary(Term); pack_value(true, _Type, _Args) -> "\"TRUE\""; pack_value(false, _Type, _Args) -> "\"FALSE\""; pack_value(Value, _Type, _Args) when is_integer(Value) -> integer_to_list(Value); pack_value(Value, _Type, _Args) when is_binary(Value) -> binary_to_list(Value); pack_value(Value, _Type, _Args) when is_float(Value) -> float_to_list(Value); pack_value(Value, _Type, _Args) when is_atom(Value) -> atom_to_list(Value); pack_value({_,_,_} = DT, _Type, _Args) -> dh_date:format("'Y-m-d H:i:s'", DT); pack_value(Value, _Type, _Args) when is_list(Value) -> "'" ++ Value ++ "'". type_to_sql(ok) -> "INTEGER"; type_to_sql(string) -> "VARCHAR"; type_to_sql(erlang_term) -> "BLOB"; type_to_sql(binary) -> "BLOB"; type_to_sql(integer) -> "INTEGER"; type_to_sql(float) -> "REAL"; type_to_sql(datetime) -> "DATETIME"; type_to_sql(timestamp) -> "TIMESTAMP"; type_to_sql(boolean) -> "ENUM(\"TRUE\",\"FALSE\")".