Permalink
Browse files

fix ensure_started to select db w/ multibulk proto, more style cleanup

  • Loading branch information...
1 parent 981147b commit 203d81d2baee3584f4c970f3541875d54b37e4e4 @japerk committed May 17, 2010
Showing with 53 additions and 42 deletions.
  1. +1 −1 ebin/erldis.app
  2. +6 −1 include/erldis.hrl
  3. +33 −28 src/erldis.erl
  4. +12 −9 src/erldis_client.erl
  5. +1 −1 src/erldis_sup.erl
  6. +0 −2 src/erldis_sync_client.erl
View
@@ -1,7 +1,7 @@
{application, erldis, [
{description, "Erlang Redis application"},
{vsn, "0.2.1"},
- {registered, [erldis_sup, erldis_client]},
+ {registered, [erldis_sup]},
{mod, {erldis_app, []}},
{applications, [kernel, stdlib]},
{modules, [
View
@@ -1 +1,6 @@
--record(redis, {socket,buffer=[],reply_caller,pipeline=false,calls=0,remaining=0,pstate=empty,results=[], host, port, timeout, db= <<"0">>, subscribers}).
+-record(redis, {
+ socket, buffer=[], reply_caller, pipeline=false, calls=0, remaining=0,
+ pstate=empty, results=[], host, port, timeout, db = <<"0">>, subscribers
+}).
+
+-define(EOL, "\r\n").
View
@@ -1,5 +1,7 @@
-module(erldis).
+-include("erldis.hrl").
+
-compile(export_all).
%%%%%%%%%%%%%%%%%%%%%%%
@@ -254,28 +256,28 @@ sort(Client, Key, Extra) when is_binary(Key), is_binary(Extra) ->
publish(Client, Channel, Value) ->
numeric(
- erldis_client:sr_scall(Client, [<<"publish">>, Channel, Value])).
+ erldis_client:sr_scall(Client, [<<"publish">>, Channel, Value])).
unsubscribe(Client)->
unsubscribe(Client, <<"">>).
unsubscribe(Client, Channel) ->
U = <<"unsubscribe">>,
Cmd = case Channel of
- <<"">> -> [U];
- _ -> [U, Channel]
- end,
+ <<"">> -> [U];
+ _ -> [U, Channel]
+ end,
case erldis_client:unsubscribe(Client, multibulk_cmd(Cmd), Channel) of
- [<<"unsubscribe">>, FirstChan, N] ->
- {FirstChan, numeric(N)};
- E ->
- E
- end.
+ [<<"unsubscribe">>, FirstChan, N] ->
+ {FirstChan, numeric(N)};
+ E ->
+ E
+ end.
subscribe(Client, Channel, Pid) ->
case erldis_client:subscribe(Client, multibulk_cmd([<<"subscribe">>, Channel]), Channel, Pid) of
- [<<"subscribe">>, Channel, N] ->
- numeric(N);
- _ ->
- error
- end.
+ [<<"subscribe">>, Channel, N] ->
+ numeric(N);
+ _ ->
+ error
+ end.
%%%%%%%%%%%%%%%%%%%%%%%%%%
%% Multiple DB commands %%
@@ -343,28 +345,31 @@ exec(Client, Fun) ->
%%%%%%%%%%%%%%%%%%%%%%%%
-define(i2l(X), integer_to_list(X)).
+
multibulk_cmd(Args) when is_binary(Args) ->
- multibulk_cmd([Args]);
+ multibulk_cmd([Args]);
multibulk_cmd(Args) when is_list(Args) ->
- TotalLength = length(Args),
-
- ArgCount = [<<"*">>, ?i2l(TotalLength), <<"\r\n">>],
- ArgBin = [[<<"$">>, ?i2l(iolist_size(A)), <<"\r\n">>,
- A, <<"\r\n">>] || A <- [erldis_binaries:to_binary(B) || B <- Args]],
-
- [ArgCount, ArgBin].
-
+ TotalLength = length(Args),
+ ArgCount = [<<"*">>, ?i2l(TotalLength), <<?EOL>>],
+ Bins = [erldis_binaries:to_binary(B) || B <- Args],
+ ArgBin = [[<<"$">>, ?i2l(iolist_size(A)), <<?EOL>>, A, <<?EOL>>] || A <- Bins],
+ [ArgCount, ArgBin].
%%%%%%%%%%%%%%%%%%%%%%
%% reply conversion %%
%%%%%%%%%%%%%%%%%%%%%%
-numeric(false) -> 0;
-numeric(true) -> 1;
-numeric(nil) -> 0;
-numeric(I) when is_binary(I) -> numeric(binary_to_list(I));
+numeric(false) ->
+ 0;
+numeric(true) ->
+ 1;
+numeric(nil) ->
+ 0;
+numeric(I) when is_binary(I) ->
+ numeric(binary_to_list(I));
numeric(I) when is_list(I) ->
- try list_to_integer(I)
+ try
+ list_to_integer(I)
catch
error:badarg ->
try list_to_float(I)
View
@@ -24,7 +24,6 @@
terminate/2, code_change/3]).
-export([subscribe/4, unsubscribe/3]).
--define(EOL, "\r\n").
-define(default_timeout, 5000). %% same as in gen.erl in stdlib
%%%%%%%%%%%%%
@@ -49,8 +48,7 @@ app_get_env(AppName, Varname, Default) ->
%%%%%%%%%%%%%%%%%%%
select(Client, DB) ->
- DBB = erldis_binaries:to_binary(DB),
- [ok] = scall(Client, <<"select ", DBB/binary>>),
+ erldis:select(Client, DB),
Client.
sr_scall(Client, Args) ->
@@ -245,8 +243,12 @@ ensure_started(#redis{socket=undefined, db=DB}=State) ->
DB == <<"0">> ->
ok;
true ->
- gen_tcp:send(Socket, <<"select ", DB/binary, ?EOL>>),
- {ok, <<"+OK", _R/binary>>} = gen_tcp:recv(Socket, 10)
+ % send & recv here since don't have an active socket
+ % because we want synchronous result since this is called
+ % from handle_* functions
+ Cmd = erldis:multibulk_cmd([<<"select">>, DB]),
+ gen_tcp:send(Socket, Cmd),
+ {ok, <<"+OK", ?EOL>>} = gen_tcp:recv(Socket, 0)
end,
inet:setopts(Socket, [{active, once}]),
@@ -271,8 +273,6 @@ connect_socket(State, _) ->
%% handle_call %%
%%%%%%%%%%%%%%%%%
-% Not sure about style here
-%
% Solves issue of remaining getting reset while still accumulating multi-bulk
% reply
dont_reset_remaining(State, Queue) ->
@@ -289,7 +289,7 @@ dont_reset_remaining(State, Queue, DB) ->
handle_call(is_pipelined, _From, State)->
{reply, State#redis.pipeline, State};
-handle_call(get_all_results, From, #redis{pipeline=true, calls=Calls} = State) ->
+handle_call(get_all_results, From, #redis{pipeline=true, calls=Calls}=State) ->
case queue:len(Calls) of
0 ->
% answers came earlier than we could start listening...
@@ -312,7 +312,10 @@ handle_call({send, Cmd}, From, State1) ->
Queue = queue:in(From, State#redis.calls),
case Cmd of
- <<"select ", DB/binary>> ->
+ % TODO: is there a cleaner way of extracting select DB command
+ % from multi-bulk commands?
+ [_, [[<<"$">>, _, <<?EOL>>, <<"select">>, <<?EOL>>],
+ [<<"$">>, _, <<?EOL>>, DB, <<?EOL>>]]] ->
{noreply, dont_reset_remaining(State, Queue, DB)};
_ ->
{noreply, dont_reset_remaining(State, Queue)}
View
@@ -19,6 +19,6 @@ init(_Args) ->
{ok, {{one_for_one, 1, 60}, [
% transient restart so client can disconnect safely
% timeout so client has time to disconnect on exit
- {erldis_client, {erldis_client, connect, []},
+ {erldis_client, {erldis_client, start_link, []},
transient, 500, worker, [erldis_client]}
]}}.
@@ -20,8 +20,6 @@
-export([init/1, handle_call/3, handle_cast/2, handle_info/2,
terminate/2, code_change/3]).
--define(EOL, "\r\n").
-
%%%%%%%%%%%%%
%% helpers %%
%%%%%%%%%%%%%

0 comments on commit 203d81d

Please sign in to comment.