client-libraries/erlang/src/client.erl

   1 -module(client).
   2 -behavior(gen_server).
   3
   4 -export([start/1, start/2, connect/1, connect/2, asend/2, send/3, send/2,
   5          disconnect/1, ssend/3, str/1, format/1, sformat/1, ssend/2,
   6          get_all_results/1]).
   7 -export([init/1, handle_call/3, handle_cast/2,
   8          handle_info/2, terminate/2, code_change/3]).
   9
  10 -include("erldis.hrl").
  11
  12 -define(EOL, "\r\n").
  13
  14
  15 %% Helpers
  16 str(X) when is_list(X) ->
  17     X;
  18 str(X) when is_atom(X) ->
  19     atom_to_list(X);
  20 str(X) when is_binary(X) ->
  21     binary_to_list(X);
  22 str(X) when is_integer(X) ->
  23     integer_to_list(X);
  24 str(X) when is_float(X) ->
  25     float_to_list(X).
  26
  27 format([], Result) ->
  28     string:join(lists:reverse(Result), ?EOL);
  29 format([Line|Rest], Result) ->
  30     JoinedLine = string:join([str(X) || X <- Line], " "),
  31     format(Rest, [JoinedLine|Result]).
  32
  33 format(Lines) ->
  34     format(Lines, []).
  35 sformat(Line) ->
  36     format([Line], []).
  37 %%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
  38
  39
  40 %% Exported API
  41 start(Host) ->
  42     connect(Host).
  43 start(Host, Port) ->
  44     connect(Host, Port).
  45
  46 connect(Host) ->
  47     connect(Host, 6379).
  48 connect(Host, Port) ->
  49     gen_server:start_link(?MODULE, [Host, Port], []).
  50
  51 % This is the simple send with a single row of commands
  52 ssend(Client, Cmd) -> ssend(Client, Cmd, []).
  53 ssend(Client, Cmd, Args) ->
  54     gen_server:cast(Client, {send, sformat([Cmd|Args])}).
  55
  56 % This is the complete send with multiple rows
  57 send(Client, Cmd) -> send(Client, Cmd, []).
  58 send(Client, Cmd, Args) ->
  59     gen_server:cast(Client, {send,
  60         string:join([str(Cmd), format(Args)], " ")}).
  61
  62 % asynchronous send, we don't care about the result.
  63 asend(Client, Cmd) ->
  64     gen_server:cast(Client, {asend, Cmd}).
  65 disconnect(Client) ->
  66     gen_server:call(Client, disconnect).
  67
  68 get_all_results(Client) ->
  69     gen_server:call(Client, get_all_results).
  70 %%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
  71
  72
  73
  74 %% gen_server callbacks
  75 init([Host, Port]) ->
  76     process_flag(trap_exit, true),
  77     ConnectOptions = [list, {active, once}, {packet, line}, {nodelay, true}],
  78     case gen_tcp:connect(Host, Port, ConnectOptions) of
  79         {error, Why} ->
  80             {error, {socket_error, Why}};
  81         {ok, Socket} ->
  82             {ok, #redis{socket=Socket, calls=0}}
  83     end.
  84
  85 handle_call({send, Cmd}, From, State) ->
  86     gen_tcp:send(State#redis.socket, [Cmd|?EOL]),
  87     {noreply, State#redis{reply_caller=fun(V) -> gen_server:reply(From, lists:nth(1, V)) end,
  88                           remaining=1}};
  89
  90 handle_call(disconnect, _From, State) ->
  91     {stop, normal, ok, State};
  92 handle_call(get_all_results, From, State) ->
  93     case State#redis.calls of
  94         0 ->
  95             % answers came earlier than we could start listening...
  96             % Very unlikely but totally possible.
  97             {reply, lists:reverse(State#redis.results), State#redis{results=[], calls=0}};
  98         _ ->
  99             % We are here earlier than results came, so just make
 100             % ourselves wait until stuff is ready.
 101             {noreply, State#redis{reply_caller=fun(V) -> gen_server:reply(From, V) end}}
 102     end;
 103 handle_call(_, _From, State) -> {noreply, State}.
 104
 105
 106 handle_cast({asend, Cmd}, State) ->
 107     gen_tcp:send(State#redis.socket, [Cmd|?EOL]),
 108     {noreply, State};
 109 handle_cast({send, Cmd}, State=#redis{remaining=Remaining, calls=Calls}) ->
 110     % how we should do here: if remaining is already != 0 then we'll
 111     % let handle_info take care of keeping track how many remaining things
 112     % there are. If instead it's 0 we are the first call so let's just
 113     % do it.
 114     gen_tcp:send(State#redis.socket, [Cmd|?EOL]),
 115     case Remaining of
 116         0 ->
 117             {noreply, State#redis{remaining=1, calls=1}};
 118         _ ->
 119             {noreply, State#redis{calls=Calls+1}}
 120     end;
 121 handle_cast(_Msg, State) -> {noreply, State}.
 122
 123
 124 trim2({ok, S}) ->
 125     string:substr(S, 1, length(S)-2);
 126 trim2(S) ->
 127     trim2({ok, S}).
 128
 129 % This function helps with pipelining by creating a pubsub system with
 130 % the caller. The caller could submit multiple requests and not listen
 131 % until later when all or some of them have been answered, at that
 132 % point 2 conditions can be true:
 133 %  1) We still need to process more things in this response chain
 134 %  2) We are finished.
 135 %
 136 % And these 2 are together with the following 2:
 137 %  1) We called get_all_results before the end of the responses.
 138 %  2) We called get_all_results after the end of the responses.
 139 %
 140 % If there's stuff missing in the chain we just push results, this also
 141 % happens when there's nothing more to process BUT we haven't requested
 142 % results yet.
 143 % In case we have requested results: if requests are not yet ready we
 144 % just push them, otherwise we finally answer all of them.
 145 save_or_reply(Result, State=#redis{calls=Calls, results=Results, reply_caller=ReplyCaller}) ->
 146     case Calls of
 147         0 ->
 148             % We don't reverse results here because if all the requests
 149             % come in and then we submit another one, if we reverse
 150             % they will be scrambled in the results field of the record.
 151             % instead if we wait just before we reply they will be
 152             % in the right order.
 153             FullResults = [Result|Results],
 154             NewState = case ReplyCaller of
 155                 undefined ->
 156                     State#redis{results=FullResults};
 157                 _ ->
 158                     ReplyCaller(lists:reverse(FullResults)),
 159                     State#redis{results=[]}
 160             end,
 161             NewState#redis{remaining=0, pstate=empty,
 162                            reply_caller=undefined, buffer=[],
 163                            calls=0};
 164         _ ->
 165             State#redis{results=[Result|Results], remaining=1, pstate=empty, buffer=[], calls=Calls}
 166
 167     end.
 168
 169 handle_info({tcp, Socket, Data}, State=#redis{calls=Calls}) ->
 170     Trimmed = trim2(Data),
 171     NewState = case {State#redis.remaining-1, proto:parse(State#redis.pstate, Trimmed)} of
 172         % This line contained an error code. Next line will hold
 173         % The error message that we will parse.
 174         {0, error} ->
 175             State#redis{remaining=1, pstate=error};
 176
 177         % The stateful parser just started and tells us the number
 178         % of results that we will have to parse for those calls
 179         % where more than one result is expected. The next
 180         % line will start with the first item to read.
 181         {0, {hold, Remaining}} ->
 182             case Remaining of
 183                 nil ->
 184                     save_or_reply(nil, State#redis{calls=Calls-1});
 185                 _ ->
 186                     % Reset the remaining value to the number of results that we need to parse.
 187                     State#redis{remaining=Remaining, pstate=read}
 188             end;
 189
 190         % We either had only one thing to read or we are at the
 191         % end of the stuff that we need to read. either way
 192         % just pack up the buffer and send.
 193         {0, {read, NBytes}} ->
 194             CurrentValue = case NBytes of
 195                 nil ->
 196                     nil;
 197                 _ ->
 198                     inet:setopts(Socket, [{packet, 0}]), % go into raw mode to read bytes
 199                     CV = trim2(gen_tcp:recv(Socket, NBytes+2)), % also consume the \r\n
 200                     inet:setopts(Socket, [{packet, line}]), % go back to line mode
 201                     CV
 202             end,
 203             OldBuffer = State#redis.buffer,
 204             case OldBuffer of
 205                 [] ->
 206                     save_or_reply(CurrentValue, State#redis{calls=Calls-1});
 207                 _ ->
 208                     save_or_reply(lists:reverse([CurrentValue|OldBuffer]), State#redis{calls=Calls-1})
 209             end;
 210
 211         % The stateful parser tells us to read some bytes
 212         {N, {read, NBytes}} ->
 213             % annoying repetition... I should reuse this code.
 214             CurrentValue = case NBytes of
 215                 nil ->
 216                     nil;
 217                 _ ->
 218                     inet:setopts(Socket, [{packet, 0}]), % go into raw mode to read bytes
 219                     CV = trim2(gen_tcp:recv(Socket, NBytes+2)), % also consume the \r\n
 220                     inet:setopts(Socket, [{packet, line}]), % go back to line mode
 221                     CV
 222             end,
 223             OldBuffer = State#redis.buffer,
 224             State#redis{remaining=N, buffer=[CurrentValue|OldBuffer], pstate=read};
 225
 226
 227         % Simple return values contained in a single line
 228         {0, Value} ->
 229             save_or_reply(Value, State#redis{calls=Calls-1})
 230
 231     end,
 232     inet:setopts(Socket, [{active, once}]),
 233     {noreply, NewState};
 234 handle_info(_Info, State) -> {noreply, State}.
 235
 236
 237 terminate(_Reason, State) ->
 238     case State#redis.socket of
 239         undefined ->
 240             pass;
 241         Socket ->
 242             gen_tcp:close(Socket)
 243     end,
 244     ok.
 245
 246
 247 code_change(_OldVsn, State, _Extra) -> {ok, State}.
 248 %%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
 249
 250