Require Redis version >= 3.2.0

Since we now use Lua scripting for cleaning up c2s sessions
the minimum supported Redis version is 3.2.0 or above because
we need to work correctly with Redis replication mechanism.

****** BACKWARD INCOMPATIBILITY WARNING *******
** THIS SHOULD BE ADDED TO THE RELEASE NOTES **
*** PACKAGE MAINTAINERS SHOULD BE INFORMED  ***
***********************************************

src/ejabberd_redis.erl

@@ -33,7 +33,7 @@
 %% API
 -export([start_link/1, get_proc/1, get_connection/1, q/1, qp/1, format_error/1]).
 %% Commands
--export([multi/1, get/1, set/2, del/1,
+-export([multi/1, get/1, set/2, del/1, info/1,
 	 sadd/2, srem/2, smembers/1, sismember/2, scard/1,
 	 hget/2, hset/3, hdel/2, hlen/1, hgetall/1, hkeys/1,
 	 subscribe/1, publish/2, script_load/1, evalsha/3]).
@@ -61,6 +61,9 @@
 -type redis_reply() :: binary() | [binary()].
 -type redis_command() :: [binary()].
 -type redis_pipeline() :: [redis_command()].
+-type redis_info() :: server | clients | memory | persistence |
+		      stats | replication | cpu | commandstats |
+		      cluster | keyspace | default | all.
 -type state() :: #state{}.
 
 -export_type([error_reason/0]).
@@ -334,6 +337,22 @@ evalsha(SHA, Keys, Args) ->
 	    erlang:error(transaction_unsupported)
     end.
 
+-spec info(redis_info()) -> {ok, [{atom(), binary()}]} | redis_error().
+info(Type) ->
+    case erlang:get(?TR_STACK) of
+	undefined ->
+	    case q([<<"INFO">>, misc:atom_to_binary(Type)]) of
+		{ok, Info} ->
+		    Lines = binary:split(Info, <<"\r\n">>, [global]),
+		    KVs = [binary:split(Line, <<":">>) || Line <- Lines],
+		    {ok, [{misc:binary_to_atom(K), V} || [K, V] <- KVs]};
+		{error, _} = Err ->
+		    Err
+	    end;
+	_ ->
+	    erlang:error(transaction_unsupported)
+    end.
+
 %%%===================================================================
 %%% gen_server callbacks
 %%%===================================================================

src/ejabberd_sm.erl

@@ -426,7 +426,11 @@ config_reloaded() ->
 init([]) ->
     process_flag(trap_exit, true),
     init_cache(),
-    lists:foreach(fun(Mod) -> Mod:init() end, get_sm_backends()),
+    case lists:foldl(
+	   fun(Mod, ok) -> Mod:init();
+	      (_, Err) -> Err
+	   end, ok, get_sm_backends()) of
+	ok ->
 	    clean_cache(),
 	    gen_iq_handler:start(?MODULE),
 	    ejabberd_hooks:add(host_up, ?MODULE, host_up, 50),
@@ -434,7 +438,10 @@ init([]) ->
 	    ejabberd_hooks:add(config_reloaded, ?MODULE, config_reloaded, 50),
 	    lists:foreach(fun host_up/1, ejabberd_config:get_myhosts()),
 	    ejabberd_commands:register_commands(get_commands_spec()),
-    {ok, #state{}}.
+	    {ok, #state{}};
+	{error, Why} ->
+	    {stop, Why}
+    end.
 
 handle_call(_Request, _From, State) ->
     Reply = ok, {reply, Reply, State}.

src/ejabberd_sm_redis.erl

@@ -40,6 +40,7 @@
 -include("logger.hrl").
 
 -define(SM_KEY, <<"ejabberd:sm">>).
+-define(MIN_REDIS_VERSION, <<"3.2.0">>).
 -record(state, {}).
 
 %%%===================================================================
@@ -142,8 +143,10 @@ get_sessions(LUser, LServer) ->
 %%%===================================================================
 init([]) ->
     ejabberd_redis:subscribe([?SM_KEY]),
-    clean_table(),
-    {ok, #state{}}.
+    case clean_table() of
+	ok -> {ok, #state{}};
+	{error, Why} -> {stop, Why}
+    end.
 
 handle_call(_Request, _From, State) ->
     Reply = ok,
@@ -240,7 +243,20 @@ clean_node_sessions(Node, Host, SHA) ->
 load_script() ->
     case misc:read_lua("redis_sm.lua") of
 	{ok, Data} ->
+	    case ejabberd_redis:info(server) of
+		{ok, Info} ->
+		    case proplists:get_value(redis_version, Info) of
+			V when V >= ?MIN_REDIS_VERSION ->
 			    ejabberd_redis:script_load(Data);
+			V ->
+			    ?CRITICAL_MSG("Unsupported Redis version: ~s. "
+					  "The version must be ~s or above",
+					  [V, ?MIN_REDIS_VERSION]),
+			    {error, unsupported_redis_version}
+		    end;
+		{error, _} = Err ->
+		    Err
+	    end;
 	{error, _} = Err ->
 	    Err
     end.

src/ejabberd_sm_sql.erl

@@ -55,7 +55,7 @@ init() ->
 		      ok;
 		  Err ->
 		      ?ERROR_MSG("failed to clean 'sm' table: ~p", [Err]),
-		      Err
+		      {error, db_failure}
 	      end;
 	 (_, Err) ->
 	      Err