Codebase list erlang-redis-client / 5c30bb0
New upstream version 1.2.0 Philipp Huebner 5 years ago
7 changed file(s) with 79 addition(s) and 56 deletion(s). Raw diff Collapse all Expand all
+0
-23
.travis.yml less more
0 language: erlang
1 notifications:
2 email: false
3 otp_release:
4 - 20.0
5 - 19.3
6 - 19.2
7 - 19.1
8 - 19.0
9 - 18.3
10 - 18.2.1
11 - 18.1
12 - 17.5
13 - 17.4
14 - 17.3
15 - 17.1
16 - 17.0
17 - R16B03-1
18 - R16B03
19 - R16B02
20 - R16B01
21 services:
22 - redis-server
+5
-1
README.md less more
2121 {ok, C} = eredis:start_link().
2222 {ok, <<"OK">>} = eredis:q(C, ["SET", "foo", "bar"]).
2323 {ok, <<"bar">>} = eredis:q(C, ["GET", "foo"]).
24
25 To connect to a Redis instance listening on a Unix domain socket:
26
27 {ok, C1} = eredis:start_link({local, "/var/run/redis.sock"}, 0).
2428
2529 MSET and MGET:
2630
113117 functions. They all include sensible defaults. `start_link/5` takes
114118 the following arguments:
115119
116 * Host, dns name or ip adress as string
120 * Host, dns name or ip adress as string; or unix domain socket as {local, Path} (available in OTP 19+)
117121 * Port, integer, default is 6379
118122 * Database, integer or 0 for default database
119123 * Password, string or empty string([]) for no password
+3
-1
src/eredis.app.src less more
22 {vsn, "1.1.0"},
33 {modules, [eredis, eredis_client, eredis_parser, eredis_sub, eredis_sub_client]},
44 {registered, []},
5 {applications, [kernel, stdlib]}
5 {applications, [kernel, stdlib]},
6 {maintainers, ["Knut Nesheim"]},
7 {licenses, ["MIT"]}
68 ]}.
+2
-1
src/eredis.erl less more
4646 start_link(Host, Port, Database, Password, ReconnectSleep, ?TIMEOUT).
4747
4848 start_link(Host, Port, Database, Password, ReconnectSleep, ConnectTimeout)
49 when is_list(Host),
49 when is_list(Host) orelse
50 (is_tuple(Host) andalso tuple_size(Host) =:= 2 andalso element(1, Host) =:= local),
5051 is_integer(Port),
5152 is_integer(Database) orelse Database == undefined,
5253 is_list(Password),
+48
-27
src/eredis_client.erl less more
7979 parser_state = eredis_parser:init(),
8080 queue = queue:new()},
8181
82 case connect(State) of
83 {ok, NewState} ->
84 {ok, NewState};
85 {error, Reason} ->
86 {stop, Reason}
82 case ReconnectSleep of
83 no_reconnect ->
84 case connect(State) of
85 {ok, _NewState} = Res -> Res;
86 {error, Reason} -> {stop, Reason}
87 end;
88 T when is_integer(T) ->
89 self() ! initiate_connection,
90 {ok, State}
8791 end.
8892
8993 handle_call({request, Req}, From, State) ->
142146 %% clients. If desired, spawn of a new process which will try to reconnect and
143147 %% notify us when Redis is ready. In the meantime, we can respond with
144148 %% an error message to all our clients.
145 handle_info({tcp_closed, _Socket}, #state{reconnect_sleep = no_reconnect,
146 queue = Queue} = State) ->
147 reply_all({error, tcp_closed}, Queue),
148 %% If we aren't going to reconnect, then there is nothing else for
149 %% this process to do.
150 {stop, normal, State#state{socket = undefined}};
151
152 handle_info({tcp_closed, _Socket}, #state{queue = Queue} = State) ->
153 Self = self(),
154 spawn(fun() -> reconnect_loop(Self, State) end),
155
156 %% tell all of our clients what has happened.
157 reply_all({error, tcp_closed}, Queue),
158
159 %% Throw away the socket and the queue, as we will never get a
160 %% response to the requests sent on the old socket. The absence of
161 %% a socket is used to signal we are "down"
162 {noreply, State#state{socket = undefined, queue = queue:new()}};
149 handle_info({tcp_closed, _Socket}, State) ->
150 maybe_reconnect(tcp_closed, State);
163151
164152 %% Redis is ready to accept requests, the given Socket is a socket
165153 %% already connected and authenticated.
170158 %% that Poolboy uses to manage the connections.
171159 handle_info(stop, State) ->
172160 {stop, shutdown, State};
161
162 handle_info(initiate_connection, #state{socket = undefined} = State) ->
163 case connect(State) of
164 {ok, NewState} ->
165 {noreply, NewState};
166 {error, Reason} ->
167 maybe_reconnect(Reason, State)
168 end;
173169
174170 handle_info(_Info, State) ->
175171 {stop, {unhandled_message, _Info}, State}.
264260 queue:in_r({N - 1, From, [Value | Replies]}, NewQueue);
265261 {empty, Queue} ->
266262 %% Oops
267 error_logger:info_msg("Nothing in queue, but got value from parser~n"),
268 throw(empty_queue)
263 error_logger:info_msg("eredis: Nothing in queue, but got value from parser~n"),
264 exit(empty_queue)
269265 end.
270266
271267 %% @doc Send `Value' to each client in queue. Only useful for sending
296292 try erlang:send(Pid, Value)
297293 catch
298294 Err:Reason ->
299 error_logger:info_msg("Failed to send message to ~p with reason ~p~n", [Pid, {Err, Reason}])
295 error_logger:info_msg("eredis: Failed to send message to ~p with reason ~p~n", [Pid, {Err, Reason}])
300296 end.
301297
302298 %% @doc: Helper for connecting to Redis, authenticating and selecting
305301 %% {SomeError, Reason}.
306302 connect(State) ->
307303 {ok, {AFamily, Addr}} = get_addr(State#state.host),
308 case gen_tcp:connect(Addr, State#state.port,
304 Port = case AFamily of
305 local -> 0;
306 _ -> State#state.port
307 end,
308 case gen_tcp:connect(Addr, Port,
309309 [AFamily | ?SOCKET_OPTS], State#state.connect_timeout) of
310310 {ok, Socket} ->
311311 case authenticate(Socket, State#state.password) of
323323 {error, {connection_error, Reason}}
324324 end.
325325
326 get_addr({local, Path}) ->
327 {ok, {local, {local, Path}}};
326328 get_addr(Hostname) ->
327329 case inet:parse_address(Hostname) of
328330 {ok, {_,_,_,_} = Addr} -> {ok, {inet, Addr}};
367369 {error, Reason} ->
368370 {error, Reason}
369371 end.
372
373 maybe_reconnect(Reason, #state{reconnect_sleep = no_reconnect, queue = Queue} = State) ->
374 reply_all({error, Reason}, Queue),
375 %% If we aren't going to reconnect, then there is nothing else for
376 %% this process to do.
377 {stop, normal, State#state{socket = undefined}};
378 maybe_reconnect(Reason, #state{queue = Queue} = State) ->
379 error_logger:error_msg("eredis: Re-establishing connection to ~p:~p due to ~p",
380 [State#state.host, State#state.port, Reason]),
381 Self = self(),
382 spawn_link(fun() -> reconnect_loop(Self, State) end),
383
384 %% tell all of our clients what has happened.
385 reply_all({error, Reason}, Queue),
386
387 %% Throw away the socket and the queue, as we will never get a
388 %% response to the requests sent on the old socket. The absence of
389 %% a socket is used to signal we are "down"
390 {noreply, State#state{socket = undefined, queue = queue:new()}}.
370391
371392 %% @doc: Loop until a connection can be established, this includes
372393 %% successfully issuing the auth and select calls. When we have a
+2
-3
src/eredis_sub_client.erl less more
179179 spawn(fun() -> reconnect_loop(Self, State) end),
180180
181181 %% Throw away the socket. The absence of a socket is used to
182 %% signal we are "down"
183 {noreply, State#state{socket = undefined}};
182 %% signal we are "down"; discard possibly patrially parsed data
183 {noreply, State#state{socket = undefined, parser_state = eredis_parser:init()}};
184184
185185 %% Controller might want to be notified about every reconnect attempt
186186 handle_info(reconnect_attempt, State) ->
353353 send_to_controller(_Msg, #state{controlling_process=undefined}) ->
354354 ok;
355355 send_to_controller(Msg, #state{controlling_process={_Ref, Pid}}) ->
356 %%error_logger:info_msg("~p ! ~p~n", [Pid, Msg]),
357356 Pid ! Msg.
+19
-0
test/eredis_tests.erl less more
150150
151151 undefined_database_test() ->
152152 ?assertMatch({ok,_}, eredis:start_link("localhost", 6379, undefined)).
153
154 connection_failure_during_start_no_reconnect_test() ->
155 process_flag(trap_exit, true),
156 Res = eredis:start_link("localhost", 6378, 0, "", no_reconnect),
157 ?assertMatch({error, _}, Res),
158 IsDied = receive {'EXIT', _, _} -> died
159 after 1000 -> still_alive end,
160 process_flag(trap_exit, false),
161 ?assertEqual(died, IsDied).
162
163 connection_failure_during_start_reconnect_test() ->
164 process_flag(trap_exit, true),
165 Res = eredis:start_link("localhost", 6378, 0, "", 100),
166 ?assertMatch({ok, _}, Res),
167 {ok, ClientPid} = Res,
168 IsDied = receive {'EXIT', ClientPid, _} -> died
169 after 400 -> still_alive end,
170 process_flag(trap_exit, false),
171 ?assertEqual(still_alive, IsDied).
153172
154173 tcp_closed_test() ->
155174 C = c(),