Merge pull request #345 from weiss/last-pep-items

Fix sending of last published PEP items to newly-available resources

src/ejabberd_c2s.erl

@@ -45,6 +45,7 @@
 	 set_aux_field/3,
 	 del_aux_field/2,
 	 get_subscription/2,
+	 send_filtered/5,
 	 broadcast/4,
 	 get_subscribed/1,
          transform_listen_option/2]).
@@ -247,6 +248,9 @@ get_subscription(LFrom, StateData) ->
        true -> none
     end.
 
+send_filtered(FsmRef, Feature, From, To, Packet) ->
+    FsmRef ! {send_filtered, Feature, From, To, Packet}.
+
 broadcast(FsmRef, Type, From, Packet) ->
     FsmRef ! {broadcast, Type, From, Packet}.
 
@@ -1738,6 +1742,26 @@ handle_info({force_update_presence, LUser}, StateName,
 		     _ -> StateData
 		   end,
     fsm_next_state(StateName, NewStateData);
+handle_info({send_filtered, Feature, From, To, Packet}, StateName, StateData) ->
+    Drop = ejabberd_hooks:run_fold(c2s_filter_packet, StateData#state.server,
+				   true, [StateData#state.server, StateData,
+					  Feature, To, Packet]),
+    NewStateData = if Drop ->
+			  ?DEBUG("Dropping packet from ~p to ~p",
+				 [jlib:jid_to_string(From),
+				  jlib:jid_to_string(To)]),
+			  StateData;
+		      true ->
+			  FinalPacket = jlib:replace_from_to(From, To, Packet),
+			  case StateData#state.jid of
+			    To ->
+				send_packet(StateData, FinalPacket);
+			    _ ->
+				ejabberd_router:route(From, To, FinalPacket),
+				StateData
+			  end
+		   end,
+    fsm_next_state(StateName, NewStateData);
 handle_info({broadcast, Type, From, Packet}, StateName, StateData) ->
     Recipients = ejabberd_hooks:run_fold(
 		   c2s_broadcast_recipients, StateData#state.server,

src/mod_caps.erl

@@ -48,7 +48,8 @@
 
 %% hook handlers
 -export([user_send_packet/3, user_receive_packet/4,
-	 c2s_presence_in/2, c2s_broadcast_recipients/6]).
+	 c2s_presence_in/2, c2s_filter_packet/6,
+	 c2s_broadcast_recipients/6]).
 
 -include("ejabberd.hrl").
 -include("logger.hrl").
@@ -266,6 +267,21 @@ c2s_presence_in(C2SState,
        true -> C2SState
     end.
 
+c2s_filter_packet(InAcc, Host, C2SState, {pep_message, Feature}, To, _Packet) ->
+    case ejabberd_c2s:get_aux_field(caps_resources, C2SState) of
+      {ok, Rs} ->
+	  LTo = jlib:jid_tolower(To),
+	  case gb_trees:lookup(LTo, Rs) of
+	    {value, Caps} ->
+		Drop = not lists:member(Feature, get_features(Host, Caps)),
+		{stop, Drop};
+	    none ->
+		{stop, true}
+	  end;
+      _ -> InAcc
+    end;
+c2s_filter_packet(Acc, _, _, _, _, _) -> Acc.
+
 c2s_broadcast_recipients(InAcc, Host, C2SState,
 			 {pep_message, Feature}, _From, _Packet) ->
     case ejabberd_c2s:get_aux_field(caps_resources,
@@ -317,6 +333,8 @@ init([Host, Opts]) ->
 		  [{max_size, MaxSize}, {life_time, LifeTime}]),
     ejabberd_hooks:add(c2s_presence_in, Host, ?MODULE,
 		       c2s_presence_in, 75),
+    ejabberd_hooks:add(c2s_filter_packet, Host, ?MODULE,
+		       c2s_filter_packet, 75),
     ejabberd_hooks:add(c2s_broadcast_recipients, Host,
 		       ?MODULE, c2s_broadcast_recipients, 75),
     ejabberd_hooks:add(user_send_packet, Host, ?MODULE,
@@ -348,6 +366,8 @@ terminate(_Reason, State) ->
     Host = State#state.host,
     ejabberd_hooks:delete(c2s_presence_in, Host, ?MODULE,
 			  c2s_presence_in, 75),
+    ejabberd_hooks:delete(c2s_filter_packet, Host, ?MODULE,
+			  c2s_filter_packet, 75),
     ejabberd_hooks:delete(c2s_broadcast_recipients, Host,
 			  ?MODULE, c2s_broadcast_recipients, 75),
     ejabberd_hooks:delete(user_send_packet, Host, ?MODULE,

src/mod_pubsub.erl

@@ -3312,7 +3312,7 @@ get_allowed_items_call(Host, NodeIdx, From, Type, Options, Owners) ->
 %%	 Number = last | integer()
 %% @doc <p>Resend the items of a node to the user.</p>
 %% @todo use cache-last-item feature
-send_items(Host, Node, NodeId, Type, {U, S, R} = LJID, last) ->
+send_items(Host, Node, NodeId, Type, LJID, last) ->
     case get_cached_item(Host, NodeId) of
       undefined ->
 	  send_items(Host, Node, NodeId, Type, LJID, 1);
@@ -3325,24 +3325,9 @@ send_items(Host, Node, NodeId, Type, {U, S, R} = LJID, last) ->
 						   children =
 						       itemsEls([LastItem])}],
 					   ModifNow, ModifUSR),
-	  case is_tuple(Host) of
+	  dispatch_items(Host, LJID, Node, Stanza)
-	    false ->
-		ejabberd_router:route(service_jid(Host),
-				      jlib:make_jid(LJID), Stanza);
-	    true ->
-		case ejabberd_sm:get_session_pid(U, S, R) of
-		  C2SPid when is_pid(C2SPid) ->
-		      ejabberd_c2s:broadcast(C2SPid,
-					     {pep_message,
-					      <<((Node))/binary, "+notify">>},
-					     _Sender = service_jid(Host),
-					     Stanza);
-		  _ -> ok
-		end
-	  end
     end;
-send_items(Host, Node, NodeId, Type, {U, S, R} = LJID,
+send_items(Host, Node, NodeId, Type, LJID, Number) ->
-	   Number) ->
     ToSend = case node_action(Host, Type, get_items,
 			      [NodeId, LJID])
 		 of
@@ -3370,22 +3355,38 @@ send_items(Host, Node, NodeId, Type, {U, S, R} = LJID,
 					attrs = nodeAttr(Node),
 					children = itemsEls(ToSend)}])
 	     end,
-    case {is_tuple(Host), Stanza} of
+    dispatch_items(Host, LJID, Node, Stanza).
-      {_, undefined} ->
+
-	  ok;
+-spec(dispatch_items/4 ::
-      {false, _} ->
+(
-	  ejabberd_router:route(service_jid(Host),
+  From   :: mod_pubsub:host(),
-				jlib:make_jid(LJID), Stanza);
+  To     :: jid(),
-      {true, _} ->
+  Node   :: mod_pubsub:nodeId(),
-	  case ejabberd_sm:get_session_pid(U, S, R) of
+  Stanza :: xmlel() | undefined)
-	    C2SPid when is_pid(C2SPid) ->
+    -> any()
-		ejabberd_c2s:broadcast(C2SPid,
+).
-				       {pep_message,
+
-					<<((Node))/binary, "+notify">>},
+dispatch_items(_From, _To, _Node, _Stanza = undefined) -> ok;
-				       _Sender = service_jid(Host), Stanza);
+dispatch_items({FromU, FromS, FromR} = From, {ToU, ToS, ToR} = To, Node,
-	    _ -> ok
+	       Stanza) ->
-	  end
+    C2SPid = case ejabberd_sm:get_session_pid(ToU, ToS, ToR) of
-    end.
+	       ToPid when is_pid(ToPid) -> ToPid;
+	       _ ->
+		   R = user_resource(FromU, FromS, FromR),
+		   case ejabberd_sm:get_session_pid(FromU, FromS, R) of
+		     FromPid when is_pid(FromPid) -> FromPid;
+		     _ -> undefined
+		   end
+	     end,
+    if C2SPid == undefined -> ok;
+       true ->
+	   ejabberd_c2s:send_filtered(C2SPid,
+				      {pep_message, <<Node/binary, "+notify">>},
+				      service_jid(From), jlib:make_jid(To),
+				      Stanza)
+    end;
+dispatch_items(From, To, _Node, Stanza) ->
+    ejabberd_router:route(service_jid(From), jlib:make_jid(To), Stanza).
 
 %% @spec (Host, JID, Plugins) -> {error, Reason} | {result, Response}
 %%	 Host = host()

src/mod_pubsub_odbc.erl

@@ -3011,8 +3011,8 @@ send_items(Host, Node, NodeId, Type, LJID, last) ->
 						       itemsEls([LastItem])}],
 					   ModifNow, ModifUSR)
     end,
-    ejabberd_router:route(service_jid(Host), jlib:make_jid(LJID), Stanza);
+    dispatch_items(Host, LJID, Node, Stanza);
-send_items(Host, Node, NodeId, Type, {U, S, R} = LJID, Number) ->
+send_items(Host, Node, NodeId, Type, LJID, Number) ->
     ToSend = case node_action(Host, Type, get_items,
 			      [NodeId, LJID])
 		 of
@@ -3040,22 +3040,38 @@ send_items(Host, Node, NodeId, Type, {U, S, R} = LJID, Number) ->
 					attrs = nodeAttr(Node),
 					children = itemsEls(ToSend)}])
 	     end,
-    case {is_tuple(Host), Stanza} of
+    dispatch_items(Host, LJID, Node, Stanza).
-      {_, undefined} ->
+
-	  ok;
+-spec(dispatch_items/4 ::
-      {false, _} ->
+(
-	  ejabberd_router:route(service_jid(Host),
+  From   :: mod_pubsub:host(),
-				jlib:make_jid(LJID), Stanza);
+  To     :: jid(),
-      {true, _} ->
+  Node   :: mod_pubsub:nodeId(),
-	  case ejabberd_sm:get_session_pid(U, S, R) of
+  Stanza :: xmlel() | undefined)
-	    C2SPid when is_pid(C2SPid) ->
+    -> any()
-		ejabberd_c2s:broadcast(C2SPid,
+).
-				       {pep_message,
+
-					<<((Node))/binary, "+notify">>},
+dispatch_items(_From, _To, _Node, _Stanza = undefined) -> ok;
-				       _Sender = service_jid(Host), Stanza);
+dispatch_items({FromU, FromS, FromR} = From, {ToU, ToS, ToR} = To, Node,
-	    _ -> ok
+	       Stanza) ->
-	  end
+    C2SPid = case ejabberd_sm:get_session_pid(ToU, ToS, ToR) of
-    end.
+	       ToPid when is_pid(ToPid) -> ToPid;
+	       _ ->
+		   R = user_resource(FromU, FromS, FromR),
+		   case ejabberd_sm:get_session_pid(FromU, FromS, R) of
+		     FromPid when is_pid(FromPid) -> FromPid;
+		     _ -> undefined
+		   end
+	     end,
+    if C2SPid == undefined -> ok;
+       true ->
+	   ejabberd_c2s:send_filtered(C2SPid,
+				      {pep_message, <<Node/binary, "+notify">>},
+				      service_jid(From), jlib:make_jid(To),
+				      Stanza)
+    end;
+dispatch_items(From, To, _Node, Stanza) ->
+    ejabberd_router:route(service_jid(From), jlib:make_jid(To), Stanza).
 
 %% @spec (Host, JID, Plugins) -> {error, Reason} | {result, Response}
 %%	 Host = host()