Copy multicast code from ejabberd-contrib to provide XEP-0033 (#521)

ejabberd.yml.example

@@ -584,6 +584,7 @@ modules:
     access_persistent: muc_create
     access_admin: muc_admin
   ## mod_muc_log: {}
+  ## mod_multicast: {}
   mod_offline: 
     access_max_user_messages: max_user_offline_messages
   mod_ping: {}

src/ejabberd_c2s.erl

@@ -2216,14 +2216,16 @@ try_roster_subscribe(Type, User, Server, From, To, Packet, StateData) ->
 presence_broadcast(StateData, From, JIDSet, Packet) ->
     JIDs = ?SETS:to_list(JIDSet),
     JIDs2 = format_and_check_privacy(From, StateData, Packet, JIDs, out),
-    send_multiple(StateData, From, JIDs2, Packet).
+    Server = StateData#state.server,
+    send_multiple(From, Server, JIDs2, Packet).
 
 %% Send presence when updating presence
 presence_broadcast_to_trusted(StateData, From, Trusted, JIDSet, Packet) ->
     JIDs = ?SETS:to_list(JIDSet),
     JIDs_trusted = [JID || JID <- JIDs, ?SETS:is_element(JID, Trusted)],
     JIDs2 = format_and_check_privacy(From, StateData, Packet, JIDs_trusted, out),
-    send_multiple(StateData, From, JIDs2, Packet).
+    Server = StateData#state.server,
+    send_multiple(From, Server, JIDs2, Packet).
 
 %% Send presence when connecting
 presence_broadcast_first(From, StateData, Packet) ->
@@ -2235,7 +2237,7 @@ presence_broadcast_first(From, StateData, Packet) ->
     PacketProbe = #xmlel{name = <<"presence">>, attrs = [{<<"type">>,<<"probe">>}], children = []},
     JIDs2Probe = format_and_check_privacy(From, StateData, PacketProbe, JIDsProbe, out),
     Server = StateData#state.server,
-    send_multiple(StateData, From, JIDs2Probe, PacketProbe),
+    send_multiple(From, Server, JIDs2Probe, PacketProbe),
     {As, JIDs} =
 	?SETS:fold(
 	   fun(JID, {A, JID_list}) ->
@@ -2244,8 +2246,7 @@ presence_broadcast_first(From, StateData, Packet) ->
 	   {StateData#state.pres_a, []},
 	   StateData#state.pres_f),
     JIDs2 = format_and_check_privacy(From, StateData, Packet, JIDs, out),
-    Server = StateData#state.server,
+    send_multiple(From, Server, JIDs2, Packet),
-    send_multiple(StateData, From, JIDs2, Packet),
     StateData#state{pres_a = As}.
 
 format_and_check_privacy(From, StateData, Packet, JIDs, Dir) ->
@@ -2266,16 +2267,8 @@ format_and_check_privacy(From, StateData, Packet, JIDs, Dir) ->
       end,
       FJIDs).
 
-send_multiple(StateData, From, JIDs, Packet) ->
+send_multiple(From, Server, JIDs, Packet) ->
-    lists:foreach(
+    ejabberd_router_multicast:route_multicast(From, Server, JIDs, Packet).
-      fun(JID) ->
-              case privacy_check_packet(StateData, From, JID, Packet, out) of
-                  deny ->
-                      ok;
-                  allow ->
-                      ejabberd_router:route(From, JID, Packet)
-              end
-      end, JIDs).
 
 remove_element(E, Set) ->
     case (?SETS):is_element(E, Set) of

src/ejabberd_router_multicast.erl

@@ -0,0 +1,239 @@
+%%%----------------------------------------------------------------------
+%%% File    : ejabberd_router_multicast.erl
+%%% Author  : Badlop <badlop@process-one.net>
+%%% Purpose : Multicast router
+%%% Created : 11 Aug 2007 by Badlop <badlop@process-one.net>
+%%%----------------------------------------------------------------------
+
+-module(ejabberd_router_multicast).
+-author('alexey@process-one.net').
+-author('badlop@process-one.net').
+
+-behaviour(gen_server).
+
+%% API
+-export([route_multicast/4,
+	 register_route/1,
+	 unregister_route/1
+	]).
+
+-export([start_link/0]).
+
+%% gen_server callbacks
+-export([init/1, handle_call/3, handle_cast/2, handle_info/2,
+	 terminate/2, code_change/3]).
+
+-include("ejabberd.hrl").
+-include("logger.hrl").
+-include("jlib.hrl").
+
+-record(route_multicast, {domain, pid}).
+-record(state, {}).
+
+%%====================================================================
+%% API
+%%====================================================================
+%%--------------------------------------------------------------------
+%% Function: start_link() -> {ok,Pid} | ignore | {error,Error}
+%% Description: Starts the server
+%%--------------------------------------------------------------------
+start_link() ->
+    gen_server:start_link({local, ?MODULE}, ?MODULE, [], []).
+
+
+route_multicast(From, Domain, Destinations, Packet) ->
+    case catch do_route(From, Domain, Destinations, Packet) of
+	{'EXIT', Reason} ->
+	    ?ERROR_MSG("~p~nwhen processing: ~p",
+		       [Reason, {From, Domain, Destinations, Packet}]);
+	_ ->
+	    ok
+    end.
+
+register_route(Domain) ->
+    case jlib:nameprep(Domain) of
+	error ->
+	    erlang:error({invalid_domain, Domain});
+	LDomain ->
+	    Pid = self(),
+	    F = fun() ->
+			mnesia:write(#route_multicast{domain = LDomain,
+						      pid = Pid})
+		end,
+	    mnesia:transaction(F)
+    end.
+
+unregister_route(Domain) ->
+    case jlib:nameprep(Domain) of
+	error ->
+	    erlang:error({invalid_domain, Domain});
+	LDomain ->
+	    Pid = self(),
+	    F = fun() ->
+		    case mnesia:select(route_multicast,
+		       [{#route_multicast{pid = Pid, domain = LDomain, _ = '_'},
+			 [],
+			 ['$_']}]) of
+			    [R] -> mnesia:delete_object(R);
+			    _ -> ok
+		    end
+		end,
+	    mnesia:transaction(F)
+    end.
+
+
+%%====================================================================
+%% gen_server callbacks
+%%====================================================================
+
+%%--------------------------------------------------------------------
+%% Function: init(Args) -> {ok, State} |
+%%                         {ok, State, Timeout} |
+%%                         ignore               |
+%%                         {stop, Reason}
+%% Description: Initiates the server
+%%--------------------------------------------------------------------
+init([]) ->
+    mnesia:create_table(route_multicast,
+			[{ram_copies, [node()]},
+			 {type, bag},
+			 {attributes,
+			  record_info(fields, route_multicast)}]),
+    mnesia:add_table_copy(route_multicast, node(), ram_copies),
+    mnesia:subscribe({table, route_multicast, simple}),
+    lists:foreach(
+      fun(Pid) ->
+	      erlang:monitor(process, Pid)
+      end,
+      mnesia:dirty_select(route_multicast, [{{route_multicast, '_', '$1'}, [], ['$1']}])),
+    {ok, #state{}}.
+
+%%--------------------------------------------------------------------
+%% 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
+%%--------------------------------------------------------------------
+handle_call(_Request, _From, State) ->
+    Reply = ok,
+    {reply, Reply, State}.
+
+%%--------------------------------------------------------------------
+%% 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({route_multicast, From, Domain, Destinations, Packet}, State) ->
+    case catch do_route(From, Domain, Destinations, Packet) of
+	{'EXIT', Reason} ->
+	    ?ERROR_MSG("~p~nwhen processing: ~p",
+		       [Reason, {From, Domain, Destinations, Packet}]);
+	_ ->
+	    ok
+    end,
+    {noreply, State};
+handle_info({mnesia_table_event, {write, #route_multicast{pid = Pid}, _ActivityId}},
+	    State) ->
+    erlang:monitor(process, Pid),
+    {noreply, State};
+handle_info({'DOWN', _Ref, _Type, Pid, _Info}, State) ->
+    F = fun() ->
+		Es = mnesia:select(
+		       route_multicast,
+		       [{#route_multicast{pid = Pid, _ = '_'},
+			 [],
+			 ['$_']}]),
+		lists:foreach(
+		  fun(E) ->
+			  mnesia:delete_object(E)
+		  end, Es)
+	end,
+    mnesia:transaction(F),
+    {noreply, State};
+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}.
+
+%%--------------------------------------------------------------------
+%%% Internal functions
+%%--------------------------------------------------------------------
+%% From = #jid
+%% Destinations = [#jid]
+do_route(From, Domain, Destinations, Packet) ->
+
+    ?DEBUG("route_multicast~n\tfrom ~s~n\tdomain ~s~n\tdestinations ~p~n\tpacket ~p~n",
+	   [jlib:jid_to_string(From),
+	    Domain,
+	    [jlib:jid_to_string(To) || To <- Destinations],
+	    Packet]),
+
+    {Groups, Rest} = lists:foldr(
+                       fun(Dest, {Groups1, Rest1}) ->
+                               case ejabberd_sm:get_session_pid(Dest#jid.luser, Dest#jid.lserver, Dest#jid.lresource) of
+                                   none ->
+                                       {Groups1, [Dest|Rest1]};
+                                   Pid ->
+                                       Node = node(Pid),
+                                       if Node /= node() ->
+                                               {dict:append(Node, Dest, Groups1), Rest1};
+                                          true ->
+                                               {Groups1, [Dest|Rest1]}
+                                       end
+                               end
+                       end, {dict:new(), []}, Destinations),
+
+    dict:map(
+      fun(Node, [Single]) ->
+              ejabberd_cluster:send({ejabberd_sm, Node},
+                                    {route, From, Single, Packet});
+         (Node, Dests) ->
+              ejabberd_cluster:send({ejabberd_sm, Node},
+                                    {route_multiple, From, Dests, Packet})
+      end, Groups),
+
+    %% Try to find an appropriate multicast service
+    case mnesia:dirty_read(route_multicast, Domain) of
+
+	%% If no multicast service is available in this server, send manually
+	[] -> do_route_normal(From, Rest, Packet);
+
+	%% If available, send the packet using multicast service
+	[R] ->
+	    case R#route_multicast.pid of
+		Pid when is_pid(Pid) ->
+		    Pid ! {route_trusted, From, Rest, Packet};
+		_ -> do_route_normal(From, Rest, Packet)
+	    end
+    end.
+
+do_route_normal(From, Destinations, Packet) ->
+    [ejabberd_router:route(From, To, Packet) || To <- Destinations].

src/ejabberd_sup.erl

@@ -62,6 +62,13 @@ init([]) ->
 	 brutal_kill,
 	 worker,
 	 [ejabberd_router]},
+    Router_multicast =
+	{ejabberd_router_multicast,
+	 {ejabberd_router_multicast, start_link, []},
+	 permanent,
+	 brutal_kill,
+	 worker,
+	 [ejabberd_router_multicast]},
     S2S =
 	{ejabberd_s2s,
 	 {ejabberd_s2s, start_link, []},
@@ -166,6 +173,7 @@ init([]) ->
 	   NodeGroups,
 	   SystemMonitor,
 	   Router,
+	   Router_multicast,
 	   S2S,
 	   Local,
 	   Captcha,

src/mod_muc_room.erl

@@ -681,14 +681,11 @@ handle_event({service_message, Msg}, _StateName,
 			children =
 			    [#xmlel{name = <<"body">>, attrs = [],
 				    children = [{xmlcdata, Msg}]}]},
-    lists:foreach(
+    send_multiple(
-      fun({_LJID, Info}) ->
+      StateData#state.jid,
-	      ejabberd_router:route(
+      StateData#state.server_host,
-		StateData#state.jid,
+      StateData#state.users,
-		Info#user.jid,
+      MessagePkt),
-		MessagePkt)
-      end,
-      ?DICT:to_list(StateData#state.users)),
     NSD = add_message_to_history(<<"">>,
 				 StateData#state.jid, MessagePkt, StateData),
     {next_state, normal_state, NSD};
@@ -945,16 +942,11 @@ process_groupchat_message(From,
 					      end,
 		 case IsAllowed of
 		   true ->
-			    lists:foreach(
+			   send_multiple(
-			      fun({_LJID, Info}) ->
+			      jlib:jid_replace_resource(StateData#state.jid, FromNick),
-				      ejabberd_router:route(
+			      StateData#state.server_host,
-					jlib:jid_replace_resource(
+			      StateData#state.users,
-					  StateData#state.jid,
+			      Packet),
-					  FromNick),
-					Info#user.jid,
-					Packet)
-			      end,
-			      ?DICT:to_list(StateData#state.users)),
 		       NewStateData2 = case has_body_or_subject(Packet) of
 			   true ->
 				add_message_to_history(FromNick, From,
@@ -4500,3 +4492,10 @@ has_body_or_subject(Packet) ->
 	(#xmlel{name = <<"subject">>}) -> false;
 	(_) -> true
     end, Packet#xmlel.children).
+
+%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
+%% Multicast
+
+send_multiple(From, Server, Users, Packet) ->
+    JIDs = [ User#user.jid || {_, User} <- ?DICT:to_list(Users)],
+    ejabberd_router_multicast:route_multicast(From, Server, JIDs, Packet).