t/merle_t01.t t/merle_t02.t @@ -8,8 +8,12 @@ docs: compile: + @mkdir -p ebin @erl -make clean: rm -f ebin/*.beam rm -f erl_crash.dump + +test: all + prove -v t/*.t Makefile @@ -24,256 +24,61 @@ %% @author Joseph Williams <joe@joetify.com> %% @copyright 2008 Joseph Williams %% @version pre 0.1 -%% @doc an erlang memcached client. +%% @seealso http://code.sixapart.com/svn/memcached/trunk/server/doc/protocol.txt +%% @doc An Erlang memcached client. %% %% This code is available as Open Source Software under the MIT license. %% %% Updates at http://github.com/joewilliams/merle/ -module(merle). +-behaviour(gen_server). -author("Joseph Williams <joe@joetify.com>"). -version("Version: pre 0.1"). --behaviour(gen_server). - -define(SERVER, ?MODULE). - --define(TCP_OPTS, [binary, {packet, raw}, {nodelay, true}, - {reuseaddr, true}, {active, true}]). +-define(TIMEOUT, 5000). +-define(TCP_OPTS, [ + binary, {packet, raw}, {nodelay, true},{reuseaddr, true}, {active, true} +]). %% gen_server API --export([start_link/2, stats/0, stats/1, version/0, - get/1, delete/2, set/4, add/4, replace/4, cas/5, quit/0]). - -%% direct API --export([stats/2, stats/3, version/2, - get/3, delete/4, set/6, add/6, replace/6, cas/7]). +-export([ + start_link/2, stats/0, stats/1, version/0, get/1, delete/2, set/4, add/4, + replace/4, cas/5 +]). %% gen_server callbacks --export([init/1, handle_call/3, handle_cast/2, handle_info/2, - terminate/2, code_change/3]). +-export([ + init/1, handle_call/3, handle_cast/2, handle_info/2, terminate/2, + code_change/3 +]). -record(state, {socket}). -%%==================================================================== -%% API -%%==================================================================== -%%-------------------------------------------------------------------- -%% Function: start_link() -> {ok,Pid} | ignore | {error,Error} -%% Description: Starts the server -%%-------------------------------------------------------------------- -start_link(Host, Port) -> - gen_server:start_link({local, ?SERVER}, ?MODULE, [Host, Port], []). - -%%==================================================================== -%% gen_server callbacks -%%==================================================================== - -%%-------------------------------------------------------------------- -%% Function: init(Args) -> {ok, State} | -%% {ok, State, Timeout} | -%% ignore | -%% {stop, Reason} -%% Description: Initiates the server -%%-------------------------------------------------------------------- -init([Host, Port]) -> - {ok, Socket} = gen_tcp:connect(Host, Port, ?TCP_OPTS), - {ok, #state{socket = Socket}}. - -%%-------------------------------------------------------------------- -%% Function: %% handle_call(Request, From, State) -> {reply, Reply, State} | -%% {reply, Reply, State, Timeout} | -%% {noreply, State} | -%% {noreply, State, Timeout} | -%% {stop, Reason, Reply, State} | -%% {stop, Reason, State} -%% Description: Handling call messages -%%-------------------------------------------------------------------- - -%% -%% Infromational Commands -%% - -handle_call({stats}, _From, #state{socket = Socket} = S) -> - Reply = send_generic_cmd(Socket, iolist_to_binary([<<"stats">>])), - {reply, Reply, S#state{socket = Socket}}; - -handle_call({stats, {Args}}, _From, #state{socket = Socket} = S) -> - Reply = send_generic_cmd(Socket, iolist_to_binary([<<"stats ">>, Args])), - {reply, Reply, S#state{socket = Socket}}; - -handle_call({version}, _From, #state{socket = Socket} = S) -> - Reply = send_generic_cmd(Socket, iolist_to_binary([<<"version">>])), - {reply, Reply, S#state{socket = Socket}}; - -%% -%% Retrieval Command -%% - -handle_call({getkey, {Key}}, _From, #state{socket = Socket} = S) -> - Reply = send_get_cmd(Socket, iolist_to_binary([<<"get ">>, Key])), - {reply, Reply, S#state{socket = Socket}}; - -%% -%% Deletion Command -%% - -handle_call({delete, {Key, Time}}, _From, #state{socket = Socket} = S) -> - Reply = send_generic_cmd(Socket, iolist_to_binary([<<"delete ">>, Key, <<" ">>, Time])), - {reply, Reply, S#state{socket = Socket}}; - -%% -%% Storage Commands -%% - -handle_call({set, {Key, Flag, ExpTime, Value}}, _From, #state{socket = Socket} = S) -> - Bin = term_to_binary(Value), - Bytes = integer_to_list(size(Bin)), - Reply = send_storage_cmd(Socket, iolist_to_binary([<<"set ">>, Key, <<" ">>, Flag, <<" ">>, - ExpTime, <<" ">>, Bytes]), Bin), - {reply, Reply, S#state{socket = Socket}}; - -handle_call({add, {Key, Flag, ExpTime, Value}}, _From, #state{socket = Socket} = S) -> - Bin = term_to_binary(Value), - Bytes = integer_to_list(size(Bin)), - Reply = send_storage_cmd(Socket, iolist_to_binary([<<"add ">>, Key, <<" ">>, Flag, <<" ">>, - ExpTime, <<" ">>, Bytes]), Bin), - {reply, Reply, S#state{socket = Socket}}; - -handle_call({replace, {Key, Flag, ExpTime, Value}}, _From, #state{socket = Socket} = S) -> - Bin = term_to_binary(Value), - Bytes = integer_to_list(size(Bin)), - Reply = send_storage_cmd(Socket, iolist_to_binary([<<"replace ">>, Key, <<" ">>, Flag, <<" ">>, - ExpTime, <<" ">>, Bytes]), Bin), - {reply, Reply, S#state{socket = Socket}}; - -handle_call({append, {Key, Value}}, _From, #state{socket = Socket} = S) -> - Bin = term_to_binary(Value), - Bytes = integer_to_list(size(Bin)), - Reply = send_storage_cmd(Socket, iolist_to_binary([<<"append ">>, Key, <<" 0 0 ">>, Bytes]), Bin), - {reply, Reply, S#state{socket = Socket}}; - -handle_call({prepend, {Key, Value}}, _From, #state{socket = Socket} = S) -> - Bin = term_to_binary(Value), - Bytes = integer_to_list(size(Bin)), - Reply = send_storage_cmd(Socket, iolist_to_binary([<<"prepend ">>, Key, <<" 0 0 ">>, Bytes]), Bin), - {reply, Reply, S#state{socket = Socket}}; - -handle_call({cas, {Key, Flag, ExpTime, CasUniq, Value}}, _From, #state{socket = Socket} = S) -> - Bin = term_to_binary(Value), - Bytes = integer_to_list(size(Bin)), - Reply = send_storage_cmd(Socket, iolist_to_binary([<<"cas ">>, Key, <<" ">>, Flag, <<" ">>, - ExpTime, <<" ">>, Bytes, <<" ">>, CasUniq]), Bin), - {reply, Reply, S#state{socket = Socket}}; - -%% -%% Exit -%% - -handle_call({quit}, _From, #state{socket = Socket} = _) -> - gen_tcp:close(Socket), - {reply, quit, {}}. - - -%%-------------------------------------------------------------------- -%% Function: handle_cast(Msg, State) -> {noreply, State} | -%% {noreply, State, Timeout} | -%% {stop, Reason, State} -%% Description: Handling cast messages -%%-------------------------------------------------------------------- -handle_cast(_Msg, State) -> - {noreply, State}. - -%%-------------------------------------------------------------------- -%% Function: handle_info(Info, State) -> {noreply, State} | -%% {noreply, State, Timeout} | -%% {stop, Reason, State} -%% Description: Handling all non call/cast messages -%%-------------------------------------------------------------------- -handle_info(_Info, State) -> - {noreply, State}. - -%%-------------------------------------------------------------------- -%% Function: terminate(Reason, State) -> void() -%% Description: 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. -%%-------------------------------------------------------------------- -terminate(_Reason, _State) -> - ok. - -%%-------------------------------------------------------------------- -%% Func: code_change(OldVsn, State, Extra) -> {ok, NewState} -%% Description: Convert process state when code is changed -%%-------------------------------------------------------------------- -code_change(_OldVsn, State, _Extra) -> - {ok, State}. - -%%-------------------------------------------------------------------- -%%% gen_server API functions -%%-------------------------------------------------------------------- - -%% Command descriptions savagely ripped from: -%% http://code.sixapart.com/svn/memcached/trunk/server/doc/protocol.txt - -%% -%% Infromational Commands -%% - %% @doc retrieve memcached stats using gen_server stats() -> gen_server:call(?SERVER, {stats}). -%% @doc retrieve memcached stats directly -stats(Host, Port) -> - Reply = send_generic_cmd(Host, Port, iolist_to_binary([<<"stats">>])), - Reply. - %% @doc retrieve memcached stats based on args using gen_server stats(Args) -> - gen_server:call(?SERVER, {stats,{Args}}). - -%% @doc retrieve memcached stats based on args directly -stats(Host, Port, Args) -> - Reply = send_generic_cmd(Host, Port, iolist_to_binary([<<"stats ">>, Args])), - Reply. + gen_server:call(?SERVER, {stats, {Args}}). %% @doc retrieve memcached version using gen_server version() -> gen_server:call(?SERVER, {version}). -%% @doc retrieve memcached stats based on args directly -version(Host, Port) -> - Reply = send_generic_cmd(Host, Port, iolist_to_binary([<<"version">>])), - Reply. - -%% -%% Retrieval Command -%% - %% @doc retrieve value based off of key using gen_server get(Key) -> - gen_server:call(?SERVER, {getkey,{Key}}). - -%% @doc retrieve value based off of key directly -get(Host, Port, Key) -> - Reply = send_get_cmd(Host, Port, iolist_to_binary([<<"get ">>, Key])), - Reply. - -%% -%% Deletion Command -%% + case gen_server:call(?SERVER, {getkey,{Key}}) of + ["END"] -> undefined; + [X] -> X + end. %% @doc delete a key and specify time using gen_server delete(Key, Time) -> gen_server:call(?SERVER, {delete, {Key, Time}}). - -%% @doc delete a key and specify time directly -delete(Host, Port, Key, Time) -> - Reply = send_generic_cmd(Host, Port, iolist_to_binary([<<"delete ">>, Key, <<" ">>, Time])), - Reply. %% Time is the amount of time in seconds %% the client wishes the server to refuse @@ -299,63 +104,139 @@ delete(Host, Port, Key, Time) -> %% @doc "store this value" using gen_server set(Key, Flag, ExpTime, Value) -> - gen_server:call(?SERVER, {set, {Key, Flag, ExpTime, Value}}). + case gen_server:call(?SERVER, {set, {Key, Flag, ExpTime, Value}}) of + ["STORED"] -> ok; + X -> X + end. -%% @doc "store this value" directly -set(Host, Port, Key, Flag, ExpTime, Value) -> - Bin = term_to_binary(Value), - Bytes = integer_to_list(size(Bin)), - Reply = send_storage_cmd(Host, Port, iolist_to_binary([<<"set ">>, Key, <<" ">>, Flag, <<" ">>, - ExpTime, <<" ">>, Bytes]), Bin), - Reply. - -%% @doc "store this value, but only if the server *doesn't* already hold Value for this key" using gen_server +%% @doc Store a key/value pair. add(Key, Flag, ExpTime, Value) -> gen_server:call(?SERVER, {add, {Key, Flag, ExpTime, Value}}). -%% @doc "store this value, but only if the server *doesn't* already hold Value for this key" directly -add(Host, Port, Key, Flag, ExpTime, Value) -> - Bin = term_to_binary(Value), - Bytes = integer_to_list(size(Bin)), - Reply = send_storage_cmd(Host, Port, iolist_to_binary([<<"add ">>, Key, <<" ">>, Flag, <<" ">>, - ExpTime, <<" ">>, Bytes]), Bin), - Reply. - -%% @doc "store this value, but only if the server *does* already hold Value for this key" using gen_server +%% @doc Replace an existing key/value pair. replace(Key, Flag, ExpTime, Value) -> gen_server:call(?SERVER, {replace, {Key, Flag, ExpTime, Value}}). -%% @doc "store this value, but only if the server *does* already hold Value for this key" directly -replace(Host, Port, Key, Flag, ExpTime, Value) -> - Bin = term_to_binary(Value), - Bytes = integer_to_list(size(Bin)), - Reply = send_storage_cmd(Host, Port, iolist_to_binary([<<"replace ">>, Key, <<" ">>, Flag, <<" ">>, - ExpTime, <<" ">>, Bytes]), Bin), - Reply. - -%% @doc "store this Vvlue but only if no one else has updated since I last fetched it" using gen_server +%% @doc Store a key/value pair if possible. cas(Key, Flag, ExpTime, CasUniq, Value) -> gen_server:call(?SERVER, {cas, {Key, Flag, ExpTime, CasUniq, Value}}). -%% @doc "store this Vvlue but only if no one else has updated since I last fetched it" directly -cas(Host, Port, Key, Flag, ExpTime, CasUniq, Value) -> +%% @private +start_link(Host, Port) -> + gen_server:start_link({local, ?SERVER}, ?MODULE, [Host, Port], []). + +%% @private +init([Host, Port]) -> + gen_tcp:connect(Host, Port, ?TCP_OPTS). + +%% @private +handle_call({stats}, _From, Socket) -> + Reply = send_generic_cmd(Socket, iolist_to_binary([<<"stats">>])), + {reply, Reply, Socket}; + +handle_call({stats, {Args}}, _From, Socket) -> + Reply = send_generic_cmd(Socket, iolist_to_binary([<<"stats ">>, Args])), + {reply, Reply, Socket}; + +handle_call({version}, _From, Socket) -> + Reply = send_generic_cmd(Socket, iolist_to_binary([<<"version">>])), + {reply, Reply, Socket}; + +handle_call({getkey, {Key}}, _From, Socket) -> + Reply = send_get_cmd(Socket, iolist_to_binary([<<"get ">>, Key])), + {reply, Reply, Socket}; + +handle_call({delete, {Key, Time}}, _From, Socket) -> + Reply = send_generic_cmd( + Socket, + iolist_to_binary([<<"delete ">>, Key, <<" ">>, Time]) + ), + {reply, Reply, Socket}; + +handle_call({set, {Key, Flag, ExpTime, Value}}, _From, Socket) -> + Bin = term_to_binary(Value), + Bytes = integer_to_list(size(Bin)), + Reply = send_storage_cmd( + Socket, + iolist_to_binary([ + <<"set ">>, Key, <<" ">>, Flag, <<" ">>, ExpTime, <<" ">>, Bytes + ]), + Bin + ), + {reply, Reply, Socket}; + +handle_call({add, {Key, Flag, ExpTime, Value}}, _From, Socket) -> Bin = term_to_binary(Value), Bytes = integer_to_list(size(Bin)), - Reply = send_storage_cmd(Host, Port, iolist_to_binary([<<"cas ">>, Key, <<" ">>, Flag, <<" ">>, - ExpTime, <<" ">>, Bytes, <<" ">>, CasUniq]), Bin), - Reply. + Reply = send_storage_cmd( + Socket, + iolist_to_binary([ + <<"add ">>, Key, <<" ">>, Flag, <<" ">>, ExpTime, <<" ">>, Bytes + ]), + Bin + ), + {reply, Reply, Socket}; + +handle_call({replace, {Key, Flag, ExpTime, Value}}, _From, Socket) -> + Bin = term_to_binary(Value), + Bytes = integer_to_list(size(Bin)), + Reply = send_storage_cmd( + Socket, + iolist_to_binary([ + <<"replace ">>, Key, <<" ">>, Flag, <<" ">>, ExpTime, <<" ">>, + Bytes + ]), + Bin + ), + {reply, Reply, Socket}; + +handle_call({append, {Key, Value}}, _From, Socket) -> + Bin = term_to_binary(Value), + Bytes = integer_to_list(size(Bin)), + Reply = send_storage_cmd( + Socket, + iolist_to_binary([<<"append ">>, Key, <<" 0 0 ">>, Bytes]), + Bin + ), + {reply, Reply, Socket}; + +handle_call({prepend, {Key, Value}}, _From, Socket) -> + Bin = term_to_binary(Value), + Bytes = integer_to_list(size(Bin)), + Reply = send_storage_cmd( + Socket, + iolist_to_binary([<<"prepend ">>, Key, <<" 0 0 ">>, Bytes]), + Bin + ), + {reply, Reply, Socket}; + +handle_call({cas, {Key, Flag, ExpTime, CasUniq, Value}}, _From, Socket) -> + Bin = term_to_binary(Value), + Bytes = integer_to_list(size(Bin)), + Reply = send_storage_cmd( + Socket, + iolist_to_binary([ + <<"cas ">>, Key, <<" ">>, Flag, <<" ">>, ExpTime, <<" ">>, Bytes, + <<" ">>, CasUniq + ]), + Bin + ), + {reply, Reply, Socket}. + +%% @private +handle_cast(_Msg, State) -> {noreply, State}. -%% -%% Exit -%% +%% @private +handle_info(_Info, State) -> {noreply, State}. -%% @doc quit() close the socket using gen_server -quit() -> - gen_server:call(?SERVER, {quit}). - -%%-------------------------------------------------------------------- -%%% Internal functions -%%-------------------------------------------------------------------- +%% @private +code_change(_OldVsn, State, _Extra) -> {ok, State}. + +%% @private +%% @doc Closes the socket +terminate(_Reason, #state{socket = Socket}) -> + gen_tcp:close(Socket), + ok. %% @private %% @doc send_generic_cmd/2 function for simple informational and deletion commands @@ -365,31 +246,12 @@ send_generic_cmd(Socket, Cmd) -> Reply. %% @private -%% @doc send_generic_cmd/3 function for simple informational and deletion commands sent directly to memcached -send_generic_cmd(Host, Port, Cmd) -> - {ok, Socket} = gen_tcp:connect(Host, Port, ?TCP_OPTS), - gen_tcp:send(Socket, <<Cmd/binary, "\r\n">>), - Reply = recv_simple_reply(), - gen_tcp:close(Socket), - Reply. - -%% @private %% @doc send_storage_cmd/3 funtion for storage commands send_storage_cmd(Socket, Cmd, Value) -> gen_tcp:send(Socket, <<Cmd/binary, "\r\n">>), gen_tcp:send(Socket, <<Value/binary, "\r\n">>), Reply = recv_simple_reply(), - Reply. - -%% @private -%% @doc send_storage_cmd/4 funtion for storage commands sent directly to memcached -send_storage_cmd(Host, Port, Cmd, Value) -> - {ok, Socket} = gen_tcp:connect(Host, Port, ?TCP_OPTS), - gen_tcp:send(Socket, <<Cmd/binary, "\r\n">>), - gen_tcp:send(Socket, <<Value/binary, "\r\n">>), - Reply = recv_simple_reply(), - gen_tcp:close(Socket), - Reply. + Reply. %% @private %% @doc send_get_cmd/2 function for retreival commands @@ -399,21 +261,12 @@ send_get_cmd(Socket, Cmd) -> Reply. %% @private -%% @doc send_get_cmd/3 function for retreival commands sent directly to memcached -send_get_cmd(Host, Port, Cmd) -> - {ok, Socket} = gen_tcp:connect(Host, Port, ?TCP_OPTS), - gen_tcp:send(Socket, <<Cmd/binary, "\r\n">>), - Reply = recv_complex_reply(Socket), - gen_tcp:close(Socket), - Reply. - -%% @private %% @doc receive function for simple responses (not containing VALUEs) recv_simple_reply() -> receive {tcp,_,Data} -> string:tokens(binary_to_list(Data), "\r\n") - after 5000 -> + after ?TIMEOUT -> timeout end. @@ -421,9 +274,9 @@ recv_simple_reply() -> %% @doc receive function for respones containing VALUEs recv_complex_reply(Socket) -> receive + %% NKG: Does this need a wilecard? %% For receiving get responses where the key does not exist - {tcp, Socket, <<"END\r\n">>} -> - ["END"]; + {tcp, Socket, <<"END\r\n">>} -> ["END"]; %% For receiving get responses containing data {tcp, Socket, Data} -> %% Reply format <<"VALUE SOMEKEY FLAG BYTES\r\nSOMEVALUE\r\nEND\r\n">> @@ -432,22 +285,20 @@ recv_complex_reply(Socket) -> Bin = list_to_binary(ListBin), Reply = get_data(Socket, Bin, Bytes, length(ListBin)), [Reply] - after 5000 -> - timeout + after ?TIMEOUT -> timeout end. %% @private -%% @doc recieve loop to get all data -get_data(Socket, Bin, Bytes, Len) -> - if - Len < Bytes + 7 -> - receive - {tcp, Socket, Data} -> - Combined = <<Bin/binary, Data/binary>>, - get_data(Socket, Combined, Bytes, size(Combined)) - after 5000 -> - timeout - end; - true -> - binary_to_term(Bin) - end. +%% @doc recieve loop to get all data +%% @todo +get_data(Socket, Bin, Bytes, Len) when Len < Bytes + 7-> + receive + %% NKG: Does this need a wildcard? + {tcp, Socket, Data} -> + Combined = <<Bin/binary, Data/binary>>, + get_data(Socket, Combined, Bytes, size(Combined)); + _ -> error + after ?TIMEOUT -> timeout + end; +get_data(_, Bin, _, _) -> + binary_to_term(Bin). src/merle.erl d75570b65b86147fbc50f4fe0f4a2901dc9033bb Nick Gerakines nick@gerakines.net http://github.com/joewilliams/merle/commit/8a13cc2b832c190f0526272b18acfe6d0bc5e7af 8a13cc2b832c190f0526272b18acfe6d0bc5e7af 2009-01-18T19:51:37-08:00 2009-01-18T19:51:37-08:00 First round of tweaks to merle. Removed a grip of redundant code, modified documentation and added etap-friendly test suite. f3c26f34604cb7ca0500cb01eea0420a7a0dcbe4 Nick Gerakines nick@gerakines.net