From 2f681b05f32ee5c5e752407a36aa667378467eae Mon Sep 17 00:00:00 2001 From: Christophe Romain Date: Tue, 8 Feb 2011 19:08:38 +0100 Subject: [PATCH] refactor pubsub broadcast to allow large optimization --- src/ejabberd_sm.erl | 12 + src/mod_pubsub/mod_pubsub.erl | 494 ++++++++++------------------ src/mod_pubsub/mod_pubsub_odbc.erl | 502 ++++++++++------------------- src/mod_pubsub/pubsub_odbc.patch | 138 ++++---- 4 files changed, 420 insertions(+), 726 deletions(-) diff --git a/src/ejabberd_sm.erl b/src/ejabberd_sm.erl index c5eb9e52d..1fd1dfa5f 100644 --- a/src/ejabberd_sm.erl +++ b/src/ejabberd_sm.erl @@ -39,6 +39,7 @@ check_in_subscription/6, bounce_offline_message/3, disconnect_removed_user/2, + get_user_sessions/2, get_user_resources/2, set_presence/5, unset_presence/4, @@ -157,6 +158,17 @@ disconnect_removed_user(User, Server) -> #xmlel{name = 'broadcast', ns = exit, attrs = [?XMLATTR(<<"reason">>, <<"User removed">>)]}). +get_user_sessions(User, Server) + when is_binary(User), is_binary(Server) -> + US = {User, Server}, + case ejabberd_cluster:get_node({User, Server}) of + Node when Node == node() -> + catch mnesia:dirty_index_read(session, US, #session.us); + Node -> + catch rpc:call(Node, mnesia, dirty_index_read, + [session, US, #session.us], 5000) + end. + get_user_resources(User, Server) when is_binary(User), is_binary(Server) -> US = {User, Server}, diff --git a/src/mod_pubsub/mod_pubsub.erl b/src/mod_pubsub/mod_pubsub.erl index d16ce38ff..c761c1f44 100644 --- a/src/mod_pubsub/mod_pubsub.erl +++ b/src/mod_pubsub/mod_pubsub.erl @@ -44,7 +44,7 @@ -module(mod_pubsub). -author('christophe.romain@process-one.net'). --version('1.13-0'). +-version('1.13-1'). -behaviour(gen_server). -behaviour(gen_mod). @@ -87,7 +87,6 @@ get_items/2, get_item/3, get_cached_item/2, - broadcast_stanza/9, get_configure/5, set_configure/5, tree_action/3, @@ -2150,10 +2149,8 @@ create_node(Host, ServerHost, Node, Owner, GivenType, Access, Configuration) -> {result, true} -> case tree_call(Host, create_node, [Host, Node, Type, Owner, NodeOptions, Parents]) of {ok, NodeId} -> - ParentTree = tree_call(Host, get_parentnodes_tree, [Host, Node, Owner]), - SubsByDepth = [{Depth, [{N, get_node_subs(N)} || N <- Nodes]} || {Depth, Nodes} <- ParentTree], case node_call(Type, create_node, [NodeId, Owner]) of - {result, Result} -> {result, {NodeId, SubsByDepth, Result}}; + {result, Result} -> {result, {NodeId, Result}}; Error -> Error end; {error, {virtual, NodeId}} -> @@ -2171,15 +2168,15 @@ create_node(Host, ServerHost, Node, Owner, GivenType, Access, Configuration) -> Reply = #xmlel{ns = ?NS_PUBSUB, name = 'pubsub', children = [#xmlel{ns = ?NS_PUBSUB, name = 'create', attrs = nodeAttr(Node)}]}, case transaction(CreateNode, transaction) of - {result, {NodeId, SubsByDepth, {Result, broadcast}}} -> - broadcast_created_node(Host, Node, NodeId, Type, NodeOptions, SubsByDepth), + {result, {NodeId, {Result, broadcast}}} -> + broadcast_created_node(Host, Node, NodeId, Type, NodeOptions), case Result of default -> {result, Reply}; _ -> {result, Result} end; - {result, {_NodeId, _SubsByDepth, default}} -> + {result, {_NodeId, default}} -> {result, Reply}; - {result, {_NodeId, _SubsByDepth, Result}} -> + {result, {_NodeId, Result}} -> {result, Result}; Error -> %% in case we change transaction to sync_dirty... @@ -2211,11 +2208,9 @@ delete_node(Host, Node, Owner) -> Action = fun(#pubsub_node{type = Type, idx = Nidx}) -> case node_call(Type, get_affiliation, [Nidx, Owner]) of {result, owner} -> - ParentTree = tree_call(Host, get_parentnodes_tree, [Host, Node, service_jid(Host)]), - SubsByDepth = [{Depth, [{N, get_node_subs(N)} || N <- Nodes]} || {Depth, Nodes} <- ParentTree], Removed = tree_call(Host, delete_node, [Host, Node]), case node_call(Type, delete_node, [Removed]) of - {result, Res} -> {result, {SubsByDepth, Res}}; + {result, Res} -> {result, Res}; Error -> Error end; _ -> @@ -2225,27 +2220,27 @@ delete_node(Host, Node, Owner) -> end, Reply = [], case transaction(Host, Node, Action, transaction) of - {result, {_, {SubsByDepth, {Result, broadcast, Removed}}}} -> + {result, {_, {Result, broadcast, Removed}}} -> lists:foreach(fun({RNode, _RSubscriptions}) -> {RH, RN} = RNode#pubsub_node.id, Nidx = RNode#pubsub_node.idx, Type = RNode#pubsub_node.type, Options = RNode#pubsub_node.options, - broadcast_removed_node(RH, RN, Nidx, Type, Options, SubsByDepth), + broadcast_removed_node(RH, RN, Nidx, Type, Options), unset_cached_item(RH, Nidx) end, Removed), case Result of default -> {result, Reply}; _ -> {result, Result} end; - {result, {_, {_, {Result, _Removed}}}} -> + {result, {_, {Result, _Removed}}} -> case Result of default -> {result, Reply}; _ -> {result, Result} end; - {result, {_, {_, default}}} -> + {result, {_, default}} -> {result, Reply}; - {result, {_, {_, Result}}} -> + {result, {_, Result}} -> {result, Result}; Error -> Error @@ -2469,12 +2464,11 @@ publish_item(Host, ServerHost, Node, Publisher, ItemId, Payload) -> Nidx = TNode#pubsub_node.idx, Type = TNode#pubsub_node.type, Options = TNode#pubsub_node.options, - BroadcastPayload = case Broadcast of - default -> Payload; + BrPayload = case Broadcast of broadcast -> Payload; PluginPayload -> PluginPayload end, - broadcast_publish_item(Host, Node, Nidx, Type, Options, Removed, ItemId, jlib:short_prepd_jid(Publisher), BroadcastPayload), + broadcast_publish_item(Host, Node, Nidx, Type, Options, ItemId, jlib:short_prepd_jid(Publisher), BrPayloadi, Removed), set_cached_item(Host, Nidx, ItemId, Publisher, Payload), case Result of default -> {result, Reply}; @@ -3315,21 +3309,20 @@ sub_to_deliver(_LJID, NotifyType, Depth, SubOptions) -> sub_option_can_deliver(NotifyType, Depth, Option) end, SubOptions). +node_to_deliver(LJID, NodeOptions) -> + presence_can_deliver(LJID, get_option(NodeOptions, presence_based_delivery)). + sub_option_can_deliver(items, _, {subscription_type, nodes}) -> false; sub_option_can_deliver(nodes, _, {subscription_type, items}) -> false; sub_option_can_deliver(_, _, {subscription_depth, all}) -> true; sub_option_can_deliver(_, Depth, {subscription_depth, D}) -> Depth =< D; -sub_option_can_deliver(_, _, {deliver, false}) -> false; -sub_option_can_deliver(_, _, {expire, When}) -> now() < When; -sub_option_can_deliver(_, _, _) -> true. - -node_to_deliver(LJID, NodeOptions) -> - PresenceDelivery = get_option(NodeOptions, presence_based_delivery), - presence_can_deliver(LJID, PresenceDelivery). +sub_option_can_deliver(_, _, {deliver, false}) -> false; +sub_option_can_deliver(_, _, {expire, When}) -> now() < When; +sub_option_can_deliver(_, _, _) -> true. presence_can_deliver(_, false) -> true; presence_can_deliver({User, Server, Resource}, true) -> - case mnesia:dirty_match_object({session, '_', '_', {User, Server}, '_', '_'}) of + case ejabberd_sm:get_user_sessions(User, Server) of [] -> false; Sessions -> lists:foldl(fun(_, true) -> true; @@ -3398,7 +3391,7 @@ payload_els_ns([#xmlel{}|Tail], Count, NS) -> payload_els_ns(Tail, Count+1, NS); payload_els_ns([_|Tail], Count, NS) -> payload_els_ns(Tail, Count, NS). %% @spec (Els) -> stanza() -%% Els = [xmlelement()] +%% Els = [xmlel()] %% @doc

Build pubsub event stanza

event_stanza(Els) -> event_stanza_withmoreels(Els, []). @@ -3413,327 +3406,174 @@ event_stanza_withmoreels(Els, MoreEls) -> #xmlel{ns = ?NS_JABBER_CLIENT, name = 'message', children = [#xmlel{ns = ?NS_PUBSUB_EVENT, name = 'event', children = Els} | MoreEls]}. +event_stanza(Event, EvAttr) -> + event_stanza([#xmlel{ns = ?NS_PUBSUB_EVENT, name = Event, attrs = EvAttr}]). +event_stanza(Event, EvAttr, Entries) -> + event_stanza([#xmlel{ns = ?NS_PUBSUB_EVENT, name = Event, attrs = EvAttr, children = + [#xmlel{ns = ?NS_PUBSUB_EVENT, name = Entry, attrs = EnAttr} || + {Entry, EnAttr} <- Entries]}]). +event_stanza(Event, EvAttr, Entry, EnAttr, Payload) -> + event_stanza([#xmlel{ns = ?NS_PUBSUB_EVENT, name = Event, attrs = EvAttr, children = + [#xmlel{ns = ?NS_PUBSUB_EVENT, name = Entry, attrs = EnAttr, children = Payload}]}]). +event_stanza(Event, EvAttr, Entry, EnAttr, Payload, Publisher) -> + Stanza = event_stanza(Event, EvAttr, Entry, EnAttr, Payload), + add_extended_headers(Stanza, extended_headers([jlib:jid_to_string(Publisher)])). + %%%%%% broadcast functions -broadcast_publish_item(Host, Node, NodeId, Type, Options, Removed, ItemId, From, Payload) -> - %broadcast(Host, Node, NodeId, Options, none, true, 'items', ItemEls) - case get_collection_subscriptions(Host, Node) of - [] -> - {result, false}; - SubsByDepth when is_list(SubsByDepth) -> - Content = case get_option(Options, deliver_payloads) of - true -> Payload; - false -> [] - end, - Stanza = event_stanza( - [#xmlel{ns = ?NS_PUBSUB_EVENT, name = 'items', attrs = nodeAttr(Node), children = - [#xmlel{ns = ?NS_PUBSUB_EVENT, name = 'item', attrs = itemAttr(ItemId), children = Content}]}]), - broadcast_stanza(Host, From, Node, NodeId, Type, Options, SubsByDepth, items, Stanza, true), - case Removed of - [] -> - ok; - _ -> - case get_option(Options, notify_retract) of - true -> - RetractStanza = event_stanza( - [#xmlel{ns = ?NS_PUBSUB_EVENT, name = 'items', attrs = nodeAttr(Node), children = - [#xmlel{ns = ?NS_PUBSUB_EVENT, name = 'retract', attrs = itemAttr(RId)} || RId <- Removed]}]), - broadcast_stanza(Host, Node, NodeId, Type, Options, SubsByDepth, items, RetractStanza, true); - _ -> - ok - end - end, - {result, true}; - _ -> - {result, false} - end. +broadcast_publish_item(Host, Node, NodeId, Type, NodeOptions, ItemId, Publisher, Payload, Removed) -> + PStanza = case get_option(NodeOptions, deliver_payloads) of + true -> event_stanza('items', nodeAttr(Node), 'item', itemAttr(ItemId), Payload, Publisher); + false -> event_stanza('items', nodeAttr(Node), 'item', itemAttr(ItemId), [], Publisher) + end, + RStanza = event_stanza('items', nodeAttr(Node), [{'retract', itemAttr(Rid)} || Rid <- Removed]), + Stanzas = [{true, PStanza, true}, {get_option(NodeOptions, notify_retract), RStanza, true}], + {result, broadcast(Host, Node, NodeId, Type, NodeOptions, items, Stanzas)}. broadcast_retract_items(Host, Node, NodeId, Type, NodeOptions, ItemIds) -> - broadcast_retract_items(Host, Node, NodeId, Type, NodeOptions, ItemIds, false). -broadcast_retract_items(_Host, _Node, _NodeId, _Type, _NodeOptions, [], _ForceNotify) -> - {result, false}; -broadcast_retract_items(Host, Node, NodeId, Type, NodeOptions, ItemIds, ForceNotify) -> - %broadcast(Host, Node, NodeId, NodeOptions, notify_retract, ForceNotify, 'retract', RetractEls) - case (get_option(NodeOptions, notify_retract) or ForceNotify) of - true -> - case get_collection_subscriptions(Host, Node) of - [] -> - {result, false}; - SubsByDepth when is_list(SubsByDepth)-> - Stanza = event_stanza( - [#xmlel{ns = ?NS_PUBSUB_EVENT, name = 'items', attrs = nodeAttr(Node), children = - [#xmlel{ns = ?NS_PUBSUB_EVENT, name = 'retract', attrs = itemAttr(ItemId)} || ItemId <- ItemIds]}]), - broadcast_stanza(Host, Node, NodeId, Type, NodeOptions, SubsByDepth, items, Stanza, true), - {result, true}; - _ -> - {result, false} - end; - _ -> - {result, false} - end. + broadcast_retract_items(Host, Node, NodeId, Type, NodeOptions, ItemIds, notify_retract). +broadcast_retract_items(_Host, _Node, _NodeId, _Type, _NodeOptions, [], _) -> + {result, false}; +broadcast_retract_items(Host, Node, NodeId, Type, NodeOptions, ItemIds, false) -> + broadcast_retract_items(Host, Node, NodeId, Type, NodeOptions, ItemIds, notify_retract); +broadcast_retract_items(Host, Node, NodeId, Type, NodeOptions, ItemIds, Notify) -> + Stanza = event_stanza('items', nodeAttr(Node), [{'retract', itemAttr(Rid)} || Rid <- ItemIds]), + {result, broadcast(Host, Node, NodeId, Type, NodeOptions, items, Notify, Stanza, true)}. broadcast_purge_node(Host, Node, NodeId, Type, NodeOptions) -> - %broadcast(Host, Node, NodeId, NodeOptions, notify_retract, false, 'purge', []) - case get_option(NodeOptions, notify_retract) of - true -> - case get_collection_subscriptions(Host, Node) of - [] -> - {result, false}; - SubsByDepth when is_list(SubsByDepth) -> - Stanza = event_stanza( - [#xmlel{ns = ?NS_PUBSUB_EVENT, name = 'purge', attrs = nodeAttr(Node)}]), - broadcast_stanza(Host, Node, NodeId, Type, NodeOptions, SubsByDepth, nodes, Stanza, false), - {result, true}; - _ -> - {result, false} - end; - _ -> - {result, false} - end. + Stanza = event_stanza('purge', nodeAttr(Node)), + {result, broadcast(Host, Node, NodeId, Type, NodeOptions, nodes, notify_retract, Stanza, false)}. -broadcast_removed_node(Host, Node, NodeId, Type, NodeOptions, SubsByDepth) -> - %broadcast(Host, Node, NodeId, NodeOptions, notify_delete, false, 'delete', []) - case get_option(NodeOptions, notify_delete) of - true -> - case SubsByDepth of - [] -> - {result, false}; - _ -> - Stanza = event_stanza( - [#xmlel{ns = ?NS_PUBSUB_EVENT, name = 'delete', attrs = nodeAttr(Node)}]), - broadcast_stanza(Host, Node, NodeId, Type, NodeOptions, SubsByDepth, nodes, Stanza, false), - {result, true} - end; - _ -> - {result, false} - end. +broadcast_removed_node(Host, Node, NodeId, Type, NodeOptions) -> + Stanza = event_stanza('delete', nodeAttr(Node)), + {result, broadcast(Host, Node, NodeId, Type, NodeOptions, nodes, notify_delete, Stanza, false)}. -broadcast_created_node(_, _, _, _, _, []) -> - {result, false}; -broadcast_created_node(Host, Node, NodeId, Type, NodeOptions, SubsByDepth) -> - Stanza = event_stanza([#xmlel{ns = ?NS_PUBSUB_EVENT, name = 'create', attrs = nodeAttr(Node)}]), - broadcast_stanza(Host, Node, NodeId, Type, NodeOptions, SubsByDepth, nodes, Stanza, true), - {result, true}. +broadcast_created_node(Host, Node, NodeId, Type, NodeOptions) -> + Stanza = event_stanza('create', nodeAttr(Node)), + {result, broadcast(Host, Node, NodeId, Type, NodeOptions, nodes, true, Stanza, true)}. broadcast_config_notification(Host, Node, NodeId, Type, NodeOptions, Lang) -> - %broadcast(Host, Node, NodeId, NodeOptions, notify_config, false, 'items', ConfigEls) - case get_option(NodeOptions, notify_config) of + Stanza = case get_option(NodeOptions, deliver_payloads) of true -> - case get_collection_subscriptions(Host, Node) of - [] -> - {result, false}; - SubsByDepth when is_list(SubsByDepth) -> - Content = case get_option(NodeOptions, deliver_payloads) of - true -> - [#xmlel{ns = ?NS_DATA_FORMS, name = 'x', attrs = [?XMLATTR(<<"type">>, <<"form">>)], children = - get_configure_xfields(Type, NodeOptions, Lang, [])}]; - false -> - [] - end, - Stanza = event_stanza( - [#xmlel{ns = ?NS_PUBSUB_EVENT, name = 'items', attrs = nodeAttr(Node), children = - [#xmlel{ns = ?NS_PUBSUB_EVENT, name = 'item', attrs = [?XMLATTR(<<"id">>, <<"configuration">>)], children = - Content}]}]), - broadcast_stanza(Host, Node, NodeId, Type, NodeOptions, SubsByDepth, nodes, Stanza, false), - {result, true}; - _ -> - {result, false} - end; - _ -> - {result, false} - end. + event_stanza([#xmlel{ns = ?NS_PUBSUB_EVENT, name = 'configuration', attrs = nodeAttr(Node), children = + [#xmlel{ns = ?NS_DATA_FORMS, name = 'x', attrs = [?XMLATTR(<<"type">>, <<"form">>)], children = + get_configure_xfields(Type, NodeOptions, Lang, [])}]}]); + false -> + event_stanza("configuration", nodeAttr(Node)) + end, + {result, broadcast(Host, Node, NodeId, Type, NodeOptions, nodes, notify_config, Stanza, false)}. +broadcast(Host, Node, NodeId, Type, NodeOptions, Notify, Stanzas) -> + Subs = node_subscriptions(Host, Node, NodeId, Type, NodeOptions, Notify), + Result = [broadcast(Host, Node, NodeId, Type, NodeOptions, Subs, Stanza, SHIM) || + {Cond, Stanza, SHIM} <- Stanzas, Cond =:= true], + lists:member(true, Result). +broadcast(Host, Node, NodeId, Type, NodeOptions, Notify, true, Stanza, SHIM) -> + Subs = node_subscriptions(Host, Node, NodeId, Type, NodeOptions, Notify), + broadcast(Host, Node, NodeId, Type, NodeOptions, Subs, Stanza, SHIM); +broadcast(_Host, _Node, _NodeId, _Type, _NodeOptions, _Notify, false, _Stanza, _SHIM) -> + false; +broadcast(Host, Node, NodeId, Type, NodeOptions, Notify, Condition, Stanza, SHIM) -> + broadcast(Host, Node, NodeId, Type, NodeOptions, Notify, get_option(NodeOptions, Condition), Stanza, SHIM). --spec(get_collection_subscriptions/2 :: - ( - Host :: host(), - NodeId :: nodeId()) - -> [] | [{Depth::integer(), Nodes :: [] | [Node::pubsubNode()]}] - ). - -get_collection_subscriptions(Host, NodeId) -> - Action = fun() -> - {result, lists:map(fun({Depth, Nodes}) -> - {Depth, [{Node, get_node_subs(Node)} || Node <- Nodes]} - end, tree_call(Host, get_parentnodes_tree, [Host, NodeId, service_jid(Host)]))} - end, - case transaction(Action, sync_dirty) of - {result, CollSubs} -> CollSubs; - _ -> [] - end. - - --spec(get_node_subs/1 :: - ( - Node::pubsubNode()) - -> [] - | [{Entity::fullUsr(), SubId::subId(), Options::[nodeOption()] | []}] - | {'error', _} - ). - -get_node_subs(#pubsub_node{type = Type, idx = NodeIdx}) -> - case node_call(Type, get_node_subscriptions, [NodeIdx]) of - {result, Subs} -> get_options_for_subs(NodeIdx, Subs); - Other -> Other - end. - - --spec(get_options_for_subs/2 :: - ( - NodeIdx :: nodeIdx(), - Subs :: [] | [{Entity::fullUsr(), Subscription::subscription(), SubId::subId()}]) - -> [] | [{Entity::fullUsr(), SubId::subId(), Options::[nodeOption()] | []}] - ). - -get_options_for_subs(NodeIdx, Subs) -> - lists:foldl(fun({Entity, 'subscribed', SubId}, Acc) -> - case pubsub_subscription:read_subscription(Entity, NodeIdx, SubId) of - {error, 'notfound'} -> [{Entity, SubId, []} | Acc]; - #pubsub_subscription{options = Options} -> [{Entity, SubId, Options} | Acc] - end; - (_, Acc) -> - Acc - end, [], Subs). - - % TODO: merge broadcast code that way - %broadcast(Host, Node, NodeId, Type, NodeOptions, Feature, Force, ElName, SubEls) -> - % case (get_option(NodeOptions, Feature) or Force) of - % true -> - % case node_action(Host, Type, get_node_subscriptions, [NodeId]) of - % {result, []} -> - % {result, false}; - % {result, Subs} -> - % Stanza = event_stanza([{xmlelement, ElName, nodeAttr(Node), SubEls}]), - % broadcast_stanza(Host, Node, Type, NodeOptions, SubOpts, Stanza), - % {result, true}; - % _ -> - % {result, false} - % end; - % _ -> - % {result, false} - % end - -broadcast_stanza(Host, _Node, _NodeId, _Type, NodeOptions, SubsByDepth, NotifyType, BaseStanza, SHIM) -> - NotificationType = get_option(NodeOptions, notification_type, headline), - BroadcastAll = get_option(NodeOptions, broadcast_all_resources), %% XXX this is not standard, but usefull - From = service_jid(Host), - Stanza = case NotificationType of - normal -> BaseStanza; - MsgType -> add_message_type(BaseStanza, atom_to_list(MsgType)) - end, - %% Handles explicit subscriptions - SubIdsByJID = subscribed_nodes_by_jid(NotifyType, SubsByDepth), - lists:foreach(fun ({LJID, NodeName, SubIds}) -> - LJIDs = case BroadcastAll of - true -> - {U, S, _} = LJID, - [{U, S, R} || R <- user_resources(U, S)]; - false -> - [LJID] - end, - %% Determine if the stanza should have SHIM ('SubId' and 'name') headers - StanzaToSend = case {SHIM, SubIds} of - {false, _} -> - Stanza; - {true, [_]} -> - add_shim_headers(Stanza, collection_shim(NodeName)); - {true, SubIds} -> - add_shim_headers(Stanza, lists:append(collection_shim(NodeName), subid_shim(SubIds))) - end, - lists:foreach(fun(To) -> - ejabberd_router:route(From, exmpp_jid:make(To), StanzaToSend) - end, LJIDs) - end, SubIdsByJID). - -broadcast_stanza({LUser, LServer, LResource}, Publisher, Node, NodeId, Type, NodeOptions, SubsByDepth, NotifyType, BaseStanza, SHIM) -> - broadcast_stanza({LUser, LServer, LResource}, Node, NodeId, Type, NodeOptions, SubsByDepth, NotifyType, BaseStanza, SHIM), - SenderResource = case LResource of - undefined -> - case user_resources(LUser, LServer) of - [Resource|_] -> Resource; - _ -> <<"">> - end; - _ -> - LResource - end, - %% Handles implicit presence subscriptions - case ejabberd_sm:get_session_pid({LUser, LServer, SenderResource}) of +broadcast({U, S, R}, Node, NodeId, Type, NodeOptions, Subscriptions, Stanza, SHIM) -> + broadcast(S, Node, NodeId, Type, NodeOptions, Subscriptions, Stanza, SHIM) + or case ejabberd_sm:get_session_pid(U, S, user_resource(U, S, R)) of C2SPid when is_pid(C2SPid) -> - Stanza = case get_option(NodeOptions, notification_type, headline) of - normal -> BaseStanza; - MsgType -> add_message_type(BaseStanza, atom_to_list(MsgType)) - end, %% set the from address on the notification to the bare JID of the account owner %% Also, add "replyto" if entity has presence subscription to the account owner %% See XEP-0163 1.1 section 4.3.1 - ejabberd_c2s:broadcast(C2SPid, - {pep_message, binary_to_list(Node)++"+notify"}, - _Sender = exmpp_jid:make(LUser, LServer), - _StanzaToSend = add_extended_headers(Stanza, - _ReplyTo = extended_headers([exmpp_jid:make(Publisher)]))); - + Event = {pep_message, binary_to_list(Node)++"+notify"}, + Message = case get_option(NodeOptions, notification_type, headline) of + normal -> Stanza; + MsgType -> add_message_type(Stanza, atom_to_list(MsgType)) + end, + ejabberd_c2s:broadcast(C2SPid, Event, jlib:make_jid(U, S, ""), Message), + true; _ -> - ?DEBUG("~p@~p has no session; can't deliver ~p to contacts", [LUser, LServer, BaseStanza]) + ?DEBUG("~p@~p has no session; can't deliver stanza: ~p", [U, S, Stanza]), + false end; -broadcast_stanza(Host, _Publisher, Node, NodeId, Type, NodeOptions, SubsByDepth, NotifyType, BaseStanza, SHIM) -> - broadcast_stanza(Host, Node, NodeId, Type, NodeOptions, SubsByDepth, NotifyType, BaseStanza, SHIM). +broadcast(_Host, _Node, _NodeId, _Type, _NodeOptions, [], _Stanza, _SHIM) -> + false; +broadcast(Host, _Node, _NodeId, _Type, NodeOptions, Subscriptions, Stanza, SHIM) -> + From = service_jid(Host), + Message = case get_option(NodeOptions, notification_type, headline) of + normal -> Stanza; + MsgType -> add_message_type(Stanza, atom_to_list(MsgType)) + end, + lists:foreach(fun({LJID, NodeName, SubIds}) -> + Send = case {SHIM, SubIds} of + {false, _} -> Message; + {true, [_]} -> add_shim_headers(Message, collection_shim(NodeName)); + {true, _} -> add_shim_headers(Message, lists:append(collection_shim(NodeName), subid_shim(SubIds))) + end, + ejabberd_router:route(From, jlib:make_jid(LJID), Send) + end, Subscriptions), + true. -subscribed_nodes_by_jid(NotifyType, SubsByDepth) -> - NodesToDeliver = fun(Depth, Node, Subs, Acc) -> - NodeName = case Node#pubsub_node.id of - {_, N} -> N; - Other -> Other - end, - NodeOptions = Node#pubsub_node.options, - lists:foldl(fun({LJID, SubId, SubOptions}, {JIDs, Recipients}) -> - case is_to_deliver(LJID, NotifyType, Depth, NodeOptions, SubOptions) of - true -> - %% If is to deliver : - case state_can_deliver(LJID, SubOptions) of - [] -> {JIDs, Recipients}; - JIDsToDeliver -> - lists:foldl( - fun(JIDToDeliver, {JIDsAcc, RecipientsAcc}) -> - case lists:member(JIDToDeliver, JIDs) of - %% check if the JIDs co-accumulator contains the Subscription JID, - false -> - %% - if not, - %% - add the JID to JIDs list co-accumulator ; - %% - create a tuple of the JID, NodeId, and SubId (as list), - %% and add the tuple to the Recipients list co-accumulator - {[JIDToDeliver | JIDsAcc], [{JIDToDeliver, NodeName, [SubId]} | RecipientsAcc]}; - true -> - %% - if the JIDs co-accumulator contains the JID - %% get the tuple containing the JID from the Recipient list co-accumulator - {_, {JIDToDeliver, NodeName1, SubIds}} = lists:keysearch(JIDToDeliver, 1, RecipientsAcc), - %% delete the tuple from the Recipients list - % v1 : Recipients1 = lists:keydelete(LJID, 1, Recipients), - % v2 : Recipients1 = lists:keyreplace(LJID, 1, Recipients, {LJID, NodeId1, [SubId | SubIds]}), - %% add the SubId to the SubIds list in the tuple, - %% and add the tuple back to the Recipients list co-accumulator - % v1.1 : {JIDs, lists:append(Recipients1, [{LJID, NodeId1, lists:append(SubIds, [SubId])}])} - % v1.2 : {JIDs, [{LJID, NodeId1, [SubId | SubIds]} | Recipients1]} - % v2: {JIDs, Recipients1} - {JIDsAcc, lists:keyreplace(JIDToDeliver, 1, RecipientsAcc, {JIDToDeliver, NodeName1, [SubId | SubIds]})} - end - end, {JIDs, Recipients}, JIDsToDeliver) - end; - false -> - {JIDs, Recipients} - end - end, Acc, Subs) - end, - DepthsToDeliver = fun({Depth, SubsByNode}, Acc1) -> - lists:foldl(fun({Node, Subs}, Acc2) -> - NodesToDeliver(Depth, Node, Subs, Acc2) - end, Acc1, SubsByNode) - end, - {_, JIDSubs} = lists:foldl(DepthsToDeliver, {[], []}, SubsByDepth), - JIDSubs. +node_subscriptions(Host, Node, NodeId, Type, _NodeOptions, Notify) -> + % TODO temporary dirty condition, should be improved using plugin or node options + case Type of + ?STDNODE -> node_subscriptions_bare(Host, Node, NodeId, Type); + ?PEPNODE -> node_subscriptions_bare(Host, Node, NodeId, Type); + _ -> node_subscriptions_full(Host, Node, Notify) + end. + +node_subscriptions_bare(Host, Node, NodeId, Type) -> + case node_action(Host, Type, get_node_subscriptions, [NodeId]) of + {result, Subs} -> + SubsByJid = lists:foldl( + fun({JID, subscribed, SubId}, Acc) -> + case dict:is_key(JID, Acc) of + true -> dict:append(JID, SubId, Acc); + false -> dict:store(JID, [SubId], Acc) + end; + (_, Acc) -> + Acc + end, dict:new(), Subs), + [{J, Node, S} || {J, S} <- dict:to_list(SubsByJid)]; + _ -> + [] + end. + +node_subscriptions_full(Host, Node, NotifyType) -> + Action = fun() -> + Collection = tree_call(Host, get_parentnodes_tree, [Host, Node, service_jid(Host)]), + {result, [{Depth, [{N, sub_with_options(N)} || N <- Nodes]} || {Depth, Nodes} <- Collection]} + end, + case transaction(Action, sync_dirty) of + {result, CollSubs} -> subscribed_nodes_by_jid(NotifyType, CollSubs); + _ -> [] + end. + +sub_with_options(#pubsub_node{type = Type, id = NodeId}) -> + case node_call(Type, get_node_subscriptions, [NodeId]) of + {result, Subs} -> + lists:foldl( + fun({JID, subscribed, SubId}, Acc) -> [sub_with_options(JID, NodeId, SubId) | Acc]; + (_, Acc) -> Acc + end, [], Subs); + _ -> + [] + end. +sub_with_options(JID, NodeId, SubId) -> + case pubsub_subscription:read_subscription(JID, NodeId, SubId) of + #pubsub_subscription{options = Options} -> {JID, SubId, Options}; + _ -> {JID, SubId, []} + end. user_resources(User, Server) -> ejabberd_sm:get_user_resources(User, Server). +user_resource(User, Server, []) -> + case user_resources(User, Server) of + [R|_] -> R; + _ -> [] + end; +user_resource(_, _, Resource) -> + Resource. + %%%%%%% Configuration handling %%

There are several reasons why the default node configuration options request might fail:

@@ -4554,7 +4394,7 @@ extended_headers(JIDs) -> on_user_offline(_, JID, _) -> {User, Server, Resource} = jlib:short_prepd_jid(JID), - case ejabberd_sm:get_user_resources(User, Server) of + case user_resources(User, Server) of [] -> purge_offline({User, Server, Resource}); _ -> true end. diff --git a/src/mod_pubsub/mod_pubsub_odbc.erl b/src/mod_pubsub/mod_pubsub_odbc.erl index 73a210710..6d8e929c7 100644 --- a/src/mod_pubsub/mod_pubsub_odbc.erl +++ b/src/mod_pubsub/mod_pubsub_odbc.erl @@ -44,7 +44,7 @@ -module(mod_pubsub_odbc). -author('christophe.romain@process-one.net'). --version('1.13-0'). +-version('1.13-1'). -behaviour(gen_server). -behaviour(gen_mod). @@ -87,7 +87,6 @@ get_items/2, get_item/3, get_cached_item/2, - broadcast_stanza/9, get_configure/5, set_configure/5, tree_action/3, @@ -684,8 +683,8 @@ disco_items(#jid{raw = JID, node = U, domain = S, resource = R} = Host, NodeId, %% ------- %% presence hooks handling functions %% -caps_update(#jid{node = U, domain = S, resource = R} = From, To, _Features) -> - Pid = ejabberd_sm:get_session_pid(U, S, R), +caps_update(From, To, _Features) -> + Pid = ejabberd_sm:get_session_pid(From), presence_probe(From, To, Pid). -spec(presence_probe/3 :: @@ -1944,10 +1943,8 @@ create_node(Host, ServerHost, Node, Owner, GivenType, Access, Configuration) -> {result, true} -> case tree_call(Host, create_node, [Host, Node, Type, Owner, NodeOptions, Parents]) of {ok, NodeId} -> - ParentTree = tree_call(Host, get_parentnodes_tree, [Host, Node, Owner]), - SubsByDepth = [{Depth, [{N, get_node_subs(N)} || N <- Nodes]} || {Depth, Nodes} <- ParentTree], case node_call(Type, create_node, [NodeId, Owner]) of - {result, Result} -> {result, {NodeId, SubsByDepth, Result}}; + {result, Result} -> {result, {NodeId, Result}}; Error -> Error end; {error, {virtual, NodeId}} -> @@ -1965,15 +1962,15 @@ create_node(Host, ServerHost, Node, Owner, GivenType, Access, Configuration) -> Reply = #xmlel{ns = ?NS_PUBSUB, name = 'pubsub', children = [#xmlel{ns = ?NS_PUBSUB, name = 'create', attrs = nodeAttr(Node)}]}, case transaction(Host, CreateNode, transaction) of - {result, {NodeId, SubsByDepth, {Result, broadcast}}} -> - broadcast_created_node(Host, Node, NodeId, Type, NodeOptions, SubsByDepth), + {result, {NodeId, {Result, broadcast}}} -> + broadcast_created_node(Host, Node, NodeId, Type, NodeOptions), case Result of default -> {result, Reply}; _ -> {result, Result} end; - {result, {_NodeId, _SubsByDepth, default}} -> + {result, {_NodeId, default}} -> {result, Reply}; - {result, {_NodeId, _SubsByDepth, Result}} -> + {result, {_NodeId, Result}} -> {result, Result}; Error -> %% in case we change transaction to sync_dirty... @@ -2005,11 +2002,9 @@ delete_node(Host, Node, Owner) -> Action = fun(#pubsub_node{type = Type, idx = Nidx}) -> case node_call(Type, get_affiliation, [Nidx, Owner]) of {result, owner} -> - ParentTree = tree_call(Host, get_parentnodes_tree, [Host, Node, service_jid(Host)]), - SubsByDepth = [{Depth, [{N, get_node_subs(N)} || N <- Nodes]} || {Depth, Nodes} <- ParentTree], Removed = tree_call(Host, delete_node, [Host, Node]), case node_call(Type, delete_node, [Removed]) of - {result, Res} -> {result, {SubsByDepth, Res}}; + {result, Res} -> {result, Res}; Error -> Error end; _ -> @@ -2019,27 +2014,27 @@ delete_node(Host, Node, Owner) -> end, Reply = [], case transaction(Host, Node, Action, transaction) of - {result, {_, {SubsByDepth, {Result, broadcast, Removed}}}} -> + {result, {_, {Result, broadcast, Removed}}} -> lists:foreach(fun({RNode, _RSubscriptions}) -> {RH, RN} = RNode#pubsub_node.id, Nidx = RNode#pubsub_node.idx, Type = RNode#pubsub_node.type, Options = RNode#pubsub_node.options, - broadcast_removed_node(RH, RN, Nidx, Type, Options, SubsByDepth), + broadcast_removed_node(RH, RN, Nidx, Type, Options), unset_cached_item(RH, Nidx) end, Removed), case Result of default -> {result, Reply}; _ -> {result, Result} end; - {result, {_, {_, {Result, _Removed}}}} -> + {result, {_, {Result, _Removed}}} -> case Result of default -> {result, Reply}; _ -> {result, Result} end; - {result, {_, {_, default}}} -> + {result, {_, default}} -> {result, Reply}; - {result, {_, {_, Result}}} -> + {result, {_, Result}} -> {result, Result}; Error -> Error @@ -2254,8 +2249,8 @@ publish_item(Host, ServerHost, Node, Publisher, ItemId, Payload) -> node_call(Type, publish_item, [Nidx, Publisher, PublishModel, MaxItems, ItemId, Payload]) end end, - ServerHostB = list_to_binary(ServerHost), - ejabberd_hooks:run(pubsub_publish_item, ServerHostB, [ServerHost, Node, Publisher, service_jid(Host), ItemId, Payload]), + %%ServerHostS = binary_to_list(ServerHost), + ejabberd_hooks:run(pubsub_publish_item, ServerHost, [ServerHost, Node, Publisher, service_jid(Host), ItemId, Payload]), Reply = #xmlel{ns = ?NS_PUBSUB, name = 'pubsub', children = [#xmlel{ns = ?NS_PUBSUB, name = 'publish', attrs = nodeAttr(Node), children = [#xmlel{ns = ?NS_PUBSUB, name = 'item', attrs = itemAttr(ItemId)}]}]}, @@ -2264,12 +2259,11 @@ publish_item(Host, ServerHost, Node, Publisher, ItemId, Payload) -> Nidx = TNode#pubsub_node.idx, Type = TNode#pubsub_node.type, Options = TNode#pubsub_node.options, - BroadcastPayload = case Broadcast of - default -> Payload; + BrPayload = case Broadcast of broadcast -> Payload; PluginPayload -> PluginPayload end, - broadcast_publish_item(Host, Node, Nidx, Type, Options, Removed, ItemId, jlib:short_prepd_jid(Publisher), BroadcastPayload), + broadcast_publish_item(Host, Node, Nidx, Type, Options, ItemId, jlib:short_prepd_jid(Publisher), BrPayloadi, Removed), set_cached_item(Host, Nidx, ItemId, Publisher, Payload), case Result of default -> {result, Reply}; @@ -3120,21 +3114,20 @@ sub_to_deliver(_LJID, NotifyType, Depth, SubOptions) -> sub_option_can_deliver(NotifyType, Depth, Option) end, SubOptions). +node_to_deliver(LJID, NodeOptions) -> + presence_can_deliver(LJID, get_option(NodeOptions, presence_based_delivery)). + sub_option_can_deliver(items, _, {subscription_type, nodes}) -> false; sub_option_can_deliver(nodes, _, {subscription_type, items}) -> false; sub_option_can_deliver(_, _, {subscription_depth, all}) -> true; sub_option_can_deliver(_, Depth, {subscription_depth, D}) -> Depth =< D; -sub_option_can_deliver(_, _, {deliver, false}) -> false; -sub_option_can_deliver(_, _, {expire, When}) -> now() < When; -sub_option_can_deliver(_, _, _) -> true. - -node_to_deliver(LJID, NodeOptions) -> - PresenceDelivery = get_option(NodeOptions, presence_based_delivery), - presence_can_deliver(LJID, PresenceDelivery). +sub_option_can_deliver(_, _, {deliver, false}) -> false; +sub_option_can_deliver(_, _, {expire, When}) -> now() < When; +sub_option_can_deliver(_, _, _) -> true. presence_can_deliver(_, false) -> true; presence_can_deliver({User, Server, Resource}, true) -> - case mnesia:dirty_match_object({session, '_', '_', {User, Server}, '_', '_'}) of + case ejabberd_sm:get_user_sessions(User, Server) of [] -> false; Sessions -> lists:foldl(fun(_, true) -> true; @@ -3203,7 +3196,7 @@ payload_els_ns([#xmlel{}|Tail], Count, NS) -> payload_els_ns(Tail, Count+1, NS); payload_els_ns([_|Tail], Count, NS) -> payload_els_ns(Tail, Count, NS). %% @spec (Els) -> stanza() -%% Els = [xmlelement()] +%% Els = [xmlel()] %% @doc

Build pubsub event stanza

event_stanza(Els) -> event_stanza_withmoreels(Els, []). @@ -3218,327 +3211,174 @@ event_stanza_withmoreels(Els, MoreEls) -> #xmlel{ns = ?NS_JABBER_CLIENT, name = 'message', children = [#xmlel{ns = ?NS_PUBSUB_EVENT, name = 'event', children = Els} | MoreEls]}. +event_stanza(Event, EvAttr) -> + event_stanza([#xmlel{ns = ?NS_PUBSUB_EVENT, name = Event, attrs = EvAttr}]). +event_stanza(Event, EvAttr, Entries) -> + event_stanza([#xmlel{ns = ?NS_PUBSUB_EVENT, name = Event, attrs = EvAttr, children = + [#xmlel{ns = ?NS_PUBSUB_EVENT, name = Entry, attrs = EnAttr} || + {Entry, EnAttr} <- Entries]}]). +event_stanza(Event, EvAttr, Entry, EnAttr, Payload) -> + event_stanza([#xmlel{ns = ?NS_PUBSUB_EVENT, name = Event, attrs = EvAttr, children = + [#xmlel{ns = ?NS_PUBSUB_EVENT, name = Entry, attrs = EnAttr, children = Payload}]}]). +event_stanza(Event, EvAttr, Entry, EnAttr, Payload, Publisher) -> + Stanza = event_stanza(Event, EvAttr, Entry, EnAttr, Payload), + add_extended_headers(Stanza, extended_headers([jlib:jid_to_string(Publisher)])). + %%%%%% broadcast functions -broadcast_publish_item(Host, Node, NodeId, Type, Options, Removed, ItemId, From, Payload) -> - %broadcast(Host, Node, NodeId, Options, none, true, 'items', ItemEls) - case get_collection_subscriptions(Host, Node) of - [] -> - {result, false}; - SubsByDepth when is_list(SubsByDepth) -> - Content = case get_option(Options, deliver_payloads) of - true -> Payload; - false -> [] - end, - Stanza = event_stanza( - [#xmlel{ns = ?NS_PUBSUB_EVENT, name = 'items', attrs = nodeAttr(Node), children = - [#xmlel{ns = ?NS_PUBSUB_EVENT, name = 'item', attrs = itemAttr(ItemId), children = Content}]}]), - broadcast_stanza(Host, From, Node, NodeId, Type, Options, SubsByDepth, items, Stanza, true), - case Removed of - [] -> - ok; - _ -> - case get_option(Options, notify_retract) of - true -> - RetractStanza = event_stanza( - [#xmlel{ns = ?NS_PUBSUB_EVENT, name = 'items', attrs = nodeAttr(Node), children = - [#xmlel{ns = ?NS_PUBSUB_EVENT, name = 'retract', attrs = itemAttr(RId)} || RId <- Removed]}]), - broadcast_stanza(Host, Node, NodeId, Type, Options, SubsByDepth, items, RetractStanza, true); - _ -> - ok - end - end, - {result, true}; - _ -> - {result, false} - end. +broadcast_publish_item(Host, Node, NodeId, Type, NodeOptions, ItemId, Publisher, Payload, Removed) -> + PStanza = case get_option(NodeOptions, deliver_payloads) of + true -> event_stanza('items', nodeAttr(Node), 'item', itemAttr(ItemId), Payload, Publisher); + false -> event_stanza('items', nodeAttr(Node), 'item', itemAttr(ItemId), [], Publisher) + end, + RStanza = event_stanza('items', nodeAttr(Node), [{'retract', itemAttr(Rid)} || Rid <- Removed]), + Stanzas = [{true, PStanza, true}, {get_option(NodeOptions, notify_retract), RStanza, true}], + {result, broadcast(Host, Node, NodeId, Type, NodeOptions, items, Stanzas)}. broadcast_retract_items(Host, Node, NodeId, Type, NodeOptions, ItemIds) -> - broadcast_retract_items(Host, Node, NodeId, Type, NodeOptions, ItemIds, false). -broadcast_retract_items(_Host, _Node, _NodeId, _Type, _NodeOptions, [], _ForceNotify) -> - {result, false}; -broadcast_retract_items(Host, Node, NodeId, Type, NodeOptions, ItemIds, ForceNotify) -> - %broadcast(Host, Node, NodeId, NodeOptions, notify_retract, ForceNotify, 'retract', RetractEls) - case (get_option(NodeOptions, notify_retract) or ForceNotify) of - true -> - case get_collection_subscriptions(Host, Node) of - [] -> - {result, false}; - SubsByDepth when is_list(SubsByDepth)-> - Stanza = event_stanza( - [#xmlel{ns = ?NS_PUBSUB_EVENT, name = 'items', attrs = nodeAttr(Node), children = - [#xmlel{ns = ?NS_PUBSUB_EVENT, name = 'retract', attrs = itemAttr(ItemId)} || ItemId <- ItemIds]}]), - broadcast_stanza(Host, Node, NodeId, Type, NodeOptions, SubsByDepth, items, Stanza, true), - {result, true}; - _ -> - {result, false} - end; - _ -> - {result, false} - end. + broadcast_retract_items(Host, Node, NodeId, Type, NodeOptions, ItemIds, notify_retract). +broadcast_retract_items(_Host, _Node, _NodeId, _Type, _NodeOptions, [], _) -> + {result, false}; +broadcast_retract_items(Host, Node, NodeId, Type, NodeOptions, ItemIds, false) -> + broadcast_retract_items(Host, Node, NodeId, Type, NodeOptions, ItemIds, notify_retract); +broadcast_retract_items(Host, Node, NodeId, Type, NodeOptions, ItemIds, Notify) -> + Stanza = event_stanza('items', nodeAttr(Node), [{'retract', itemAttr(Rid)} || Rid <- ItemIds]), + {result, broadcast(Host, Node, NodeId, Type, NodeOptions, items, Notify, Stanza, true)}. broadcast_purge_node(Host, Node, NodeId, Type, NodeOptions) -> - %broadcast(Host, Node, NodeId, NodeOptions, notify_retract, false, 'purge', []) - case get_option(NodeOptions, notify_retract) of - true -> - case get_collection_subscriptions(Host, Node) of - [] -> - {result, false}; - SubsByDepth when is_list(SubsByDepth) -> - Stanza = event_stanza( - [#xmlel{ns = ?NS_PUBSUB_EVENT, name = 'purge', attrs = nodeAttr(Node)}]), - broadcast_stanza(Host, Node, NodeId, Type, NodeOptions, SubsByDepth, nodes, Stanza, false), - {result, true}; - _ -> - {result, false} - end; - _ -> - {result, false} - end. + Stanza = event_stanza('purge', nodeAttr(Node)), + {result, broadcast(Host, Node, NodeId, Type, NodeOptions, nodes, notify_retract, Stanza, false)}. -broadcast_removed_node(Host, Node, NodeId, Type, NodeOptions, SubsByDepth) -> - %broadcast(Host, Node, NodeId, NodeOptions, notify_delete, false, 'delete', []) - case get_option(NodeOptions, notify_delete) of - true -> - case SubsByDepth of - [] -> - {result, false}; - _ -> - Stanza = event_stanza( - [#xmlel{ns = ?NS_PUBSUB_EVENT, name = 'delete', attrs = nodeAttr(Node)}]), - broadcast_stanza(Host, Node, NodeId, Type, NodeOptions, SubsByDepth, nodes, Stanza, false), - {result, true} - end; - _ -> - {result, false} - end. +broadcast_removed_node(Host, Node, NodeId, Type, NodeOptions) -> + Stanza = event_stanza('delete', nodeAttr(Node)), + {result, broadcast(Host, Node, NodeId, Type, NodeOptions, nodes, notify_delete, Stanza, false)}. -broadcast_created_node(_, _, _, _, _, []) -> - {result, false}; -broadcast_created_node(Host, Node, NodeId, Type, NodeOptions, SubsByDepth) -> - Stanza = event_stanza([#xmlel{ns = ?NS_PUBSUB_EVENT, name = 'create', attrs = nodeAttr(Node)}]), - broadcast_stanza(Host, Node, NodeId, Type, NodeOptions, SubsByDepth, nodes, Stanza, true), - {result, true}. +broadcast_created_node(Host, Node, NodeId, Type, NodeOptions) -> + Stanza = event_stanza('create', nodeAttr(Node)), + {result, broadcast(Host, Node, NodeId, Type, NodeOptions, nodes, true, Stanza, true)}. broadcast_config_notification(Host, Node, NodeId, Type, NodeOptions, Lang) -> - %broadcast(Host, Node, NodeId, NodeOptions, notify_config, false, 'items', ConfigEls) - case get_option(NodeOptions, notify_config) of + Stanza = case get_option(NodeOptions, deliver_payloads) of true -> - case get_collection_subscriptions(Host, Node) of - [] -> - {result, false}; - SubsByDepth when is_list(SubsByDepth) -> - Content = case get_option(NodeOptions, deliver_payloads) of - true -> - [#xmlel{ns = ?NS_DATA_FORMS, name = 'x', attrs = [?XMLATTR(<<"type">>, <<"form">>)], children = - get_configure_xfields(Type, NodeOptions, Lang, [])}]; - false -> - [] - end, - Stanza = event_stanza( - [#xmlel{ns = ?NS_PUBSUB_EVENT, name = 'items', attrs = nodeAttr(Node), children = - [#xmlel{ns = ?NS_PUBSUB_EVENT, name = 'item', attrs = [?XMLATTR(<<"id">>, <<"configuration">>)], children = - Content}]}]), - broadcast_stanza(Host, Node, NodeId, Type, NodeOptions, SubsByDepth, nodes, Stanza, false), - {result, true}; - _ -> - {result, false} - end; - _ -> - {result, false} - end. + event_stanza([#xmlel{ns = ?NS_PUBSUB_EVENT, name = 'configuration', attrs = nodeAttr(Node), children = + [#xmlel{ns = ?NS_DATA_FORMS, name = 'x', attrs = [?XMLATTR(<<"type">>, <<"form">>)], children = + get_configure_xfields(Type, NodeOptions, Lang, [])}]}]); + false -> + event_stanza("configuration", nodeAttr(Node)) + end, + {result, broadcast(Host, Node, NodeId, Type, NodeOptions, nodes, notify_config, Stanza, false)}. +broadcast(Host, Node, NodeId, Type, NodeOptions, Notify, Stanzas) -> + Subs = node_subscriptions(Host, Node, NodeId, Type, NodeOptions, Notify), + Result = [broadcast(Host, Node, NodeId, Type, NodeOptions, Subs, Stanza, SHIM) || + {Cond, Stanza, SHIM} <- Stanzas, Cond =:= true], + lists:member(true, Result). +broadcast(Host, Node, NodeId, Type, NodeOptions, Notify, true, Stanza, SHIM) -> + Subs = node_subscriptions(Host, Node, NodeId, Type, NodeOptions, Notify), + broadcast(Host, Node, NodeId, Type, NodeOptions, Subs, Stanza, SHIM); +broadcast(_Host, _Node, _NodeId, _Type, _NodeOptions, _Notify, false, _Stanza, _SHIM) -> + false; +broadcast(Host, Node, NodeId, Type, NodeOptions, Notify, Condition, Stanza, SHIM) -> + broadcast(Host, Node, NodeId, Type, NodeOptions, Notify, get_option(NodeOptions, Condition), Stanza, SHIM). --spec(get_collection_subscriptions/2 :: - ( - Host :: host(), - NodeId :: nodeId()) - -> [] | [{Depth::integer(), Nodes :: [] | [Node::pubsubNode()]}] - ). - -get_collection_subscriptions(Host, NodeId) -> - Action = fun() -> - {result, lists:map(fun({Depth, Nodes}) -> - {Depth, [{Node, get_node_subs(Node)} || Node <- Nodes]} - end, tree_call(Host, get_parentnodes_tree, [Host, NodeId, service_jid(Host)]))} - end, - case transaction(Host, Action, sync_dirty) of - {result, CollSubs} -> CollSubs; - _ -> [] - end. - - --spec(get_node_subs/1 :: - ( - Node::pubsubNode()) - -> [] - | [{Entity::fullUsr(), SubId::subId(), Options::[nodeOption()] | []}] - | {'error', _} - ). - -get_node_subs(#pubsub_node{type = Type, idx = NodeIdx}) -> - case node_call(Type, get_node_subscriptions, [NodeIdx]) of - {result, Subs} -> get_options_for_subs(NodeIdx, Subs); - Other -> Other - end. - - --spec(get_options_for_subs/2 :: - ( - NodeIdx :: nodeIdx(), - Subs :: [] | [{Entity::fullUsr(), Subscription::subscription(), SubId::subId()}]) - -> [] | [{Entity::fullUsr(), SubId::subId(), Options::[nodeOption()] | []}] - ). - -get_options_for_subs(NodeIdx, Subs) -> - lists:foldl(fun({Entity, 'subscribed', SubId}, Acc) -> - case pubsub_subscription_odbc:read_subscription(Entity, NodeIdx, SubId) of - {error, 'notfound'} -> [{Entity, SubId, []} | Acc]; - #pubsub_subscription{options = Options} -> [{Entity, SubId, Options} | Acc] - end; - (_, Acc) -> - Acc - end, [], Subs). - - % TODO: merge broadcast code that way - %broadcast(Host, Node, NodeId, Type, NodeOptions, Feature, Force, ElName, SubEls) -> - % case (get_option(NodeOptions, Feature) or Force) of - % true -> - % case node_action(Host, Type, get_node_subscriptions, [NodeId]) of - % {result, []} -> - % {result, false}; - % {result, Subs} -> - % Stanza = event_stanza([{xmlelement, ElName, nodeAttr(Node), SubEls}]), - % broadcast_stanza(Host, Node, Type, NodeOptions, SubOpts, Stanza), - % {result, true}; - % _ -> - % {result, false} - % end; - % _ -> - % {result, false} - % end - -broadcast_stanza(Host, _Node, _NodeId, _Type, NodeOptions, SubsByDepth, NotifyType, BaseStanza, SHIM) -> - NotificationType = get_option(NodeOptions, notification_type, headline), - BroadcastAll = get_option(NodeOptions, broadcast_all_resources), %% XXX this is not standard, but usefull - From = service_jid(Host), - Stanza = case NotificationType of - normal -> BaseStanza; - MsgType -> add_message_type(BaseStanza, atom_to_list(MsgType)) - end, - %% Handles explicit subscriptions - SubIdsByJID = subscribed_nodes_by_jid(NotifyType, SubsByDepth), - lists:foreach(fun ({LJID, NodeName, SubIds}) -> - LJIDs = case BroadcastAll of - true -> - {U, S, _} = LJID, - [{U, S, R} || R <- user_resources(U, S)]; - false -> - [LJID] - end, - %% Determine if the stanza should have SHIM ('SubId' and 'name') headers - StanzaToSend = case {SHIM, SubIds} of - {false, _} -> - Stanza; - {true, [_]} -> - add_shim_headers(Stanza, collection_shim(NodeName)); - {true, SubIds} -> - add_shim_headers(Stanza, lists:append(collection_shim(NodeName), subid_shim(SubIds))) - end, - lists:foreach(fun(To) -> - ejabberd_router:route(From, exmpp_jid:make(To), StanzaToSend) - end, LJIDs) - end, SubIdsByJID). - -broadcast_stanza({LUser, LServer, LResource}, Publisher, Node, NodeId, Type, NodeOptions, SubsByDepth, NotifyType, BaseStanza, SHIM) -> - broadcast_stanza({LUser, LServer, LResource}, Node, NodeId, Type, NodeOptions, SubsByDepth, NotifyType, BaseStanza, SHIM), - SenderResource = case LResource of - undefined -> - case user_resources(LUser, LServer) of - [Resource|_] -> Resource; - _ -> <<"">> - end; - _ -> - LResource - end, - %% Handles implicit presence subscriptions - case ejabberd_sm:get_session_pid({LUser, LServer, SenderResource}) of +broadcast({U, S, R}, Node, NodeId, Type, NodeOptions, Subscriptions, Stanza, SHIM) -> + broadcast(S, Node, NodeId, Type, NodeOptions, Subscriptions, Stanza, SHIM) + or case ejabberd_sm:get_session_pid(U, S, user_resource(U, S, R)) of C2SPid when is_pid(C2SPid) -> - Stanza = case get_option(NodeOptions, notification_type, headline) of - normal -> BaseStanza; - MsgType -> add_message_type(BaseStanza, atom_to_list(MsgType)) - end, %% set the from address on the notification to the bare JID of the account owner %% Also, add "replyto" if entity has presence subscription to the account owner %% See XEP-0163 1.1 section 4.3.1 - ejabberd_c2s:broadcast(C2SPid, - {pep_message, binary_to_list(Node)++"+notify"}, - _Sender = exmpp_jid:make(LUser, LServer), - _StanzaToSend = add_extended_headers(Stanza, - _ReplyTo = extended_headers([exmpp_jid:make(Publisher)]))); - + Event = {pep_message, binary_to_list(Node)++"+notify"}, + Message = case get_option(NodeOptions, notification_type, headline) of + normal -> Stanza; + MsgType -> add_message_type(Stanza, atom_to_list(MsgType)) + end, + ejabberd_c2s:broadcast(C2SPid, Event, jlib:make_jid(U, S, ""), Message), + true; _ -> - ?DEBUG("~p@~p has no session; can't deliver ~p to contacts", [LUser, LServer, BaseStanza]) + ?DEBUG("~p@~p has no session; can't deliver stanza: ~p", [U, S, Stanza]), + false end; -broadcast_stanza(Host, _Publisher, Node, NodeId, Type, NodeOptions, SubsByDepth, NotifyType, BaseStanza, SHIM) -> - broadcast_stanza(Host, Node, NodeId, Type, NodeOptions, SubsByDepth, NotifyType, BaseStanza, SHIM). +broadcast(_Host, _Node, _NodeId, _Type, _NodeOptions, [], _Stanza, _SHIM) -> + false; +broadcast(Host, _Node, _NodeId, _Type, NodeOptions, Subscriptions, Stanza, SHIM) -> + From = service_jid(Host), + Message = case get_option(NodeOptions, notification_type, headline) of + normal -> Stanza; + MsgType -> add_message_type(Stanza, atom_to_list(MsgType)) + end, + lists:foreach(fun({LJID, NodeName, SubIds}) -> + Send = case {SHIM, SubIds} of + {false, _} -> Message; + {true, [_]} -> add_shim_headers(Message, collection_shim(NodeName)); + {true, _} -> add_shim_headers(Message, lists:append(collection_shim(NodeName), subid_shim(SubIds))) + end, + ejabberd_router:route(From, jlib:make_jid(LJID), Send) + end, Subscriptions), + true. -subscribed_nodes_by_jid(NotifyType, SubsByDepth) -> - NodesToDeliver = fun(Depth, Node, Subs, Acc) -> - NodeName = case Node#pubsub_node.id of - {_, N} -> N; - Other -> Other - end, - NodeOptions = Node#pubsub_node.options, - lists:foldl(fun({LJID, SubId, SubOptions}, {JIDs, Recipients}) -> - case is_to_deliver(LJID, NotifyType, Depth, NodeOptions, SubOptions) of - true -> - %% If is to deliver : - case state_can_deliver(LJID, SubOptions) of - [] -> {JIDs, Recipients}; - JIDsToDeliver -> - lists:foldl( - fun(JIDToDeliver, {JIDsAcc, RecipientsAcc}) -> - case lists:member(JIDToDeliver, JIDs) of - %% check if the JIDs co-accumulator contains the Subscription JID, - false -> - %% - if not, - %% - add the JID to JIDs list co-accumulator ; - %% - create a tuple of the JID, NodeId, and SubId (as list), - %% and add the tuple to the Recipients list co-accumulator - {[JIDToDeliver | JIDsAcc], [{JIDToDeliver, NodeName, [SubId]} | RecipientsAcc]}; - true -> - %% - if the JIDs co-accumulator contains the JID - %% get the tuple containing the JID from the Recipient list co-accumulator - {_, {JIDToDeliver, NodeName1, SubIds}} = lists:keysearch(JIDToDeliver, 1, RecipientsAcc), - %% delete the tuple from the Recipients list - % v1 : Recipients1 = lists:keydelete(LJID, 1, Recipients), - % v2 : Recipients1 = lists:keyreplace(LJID, 1, Recipients, {LJID, NodeId1, [SubId | SubIds]}), - %% add the SubId to the SubIds list in the tuple, - %% and add the tuple back to the Recipients list co-accumulator - % v1.1 : {JIDs, lists:append(Recipients1, [{LJID, NodeId1, lists:append(SubIds, [SubId])}])} - % v1.2 : {JIDs, [{LJID, NodeId1, [SubId | SubIds]} | Recipients1]} - % v2: {JIDs, Recipients1} - {JIDsAcc, lists:keyreplace(JIDToDeliver, 1, RecipientsAcc, {JIDToDeliver, NodeName1, [SubId | SubIds]})} - end - end, {JIDs, Recipients}, JIDsToDeliver) - end; - false -> - {JIDs, Recipients} - end - end, Acc, Subs) - end, - DepthsToDeliver = fun({Depth, SubsByNode}, Acc1) -> - lists:foldl(fun({Node, Subs}, Acc2) -> - NodesToDeliver(Depth, Node, Subs, Acc2) - end, Acc1, SubsByNode) - end, - {_, JIDSubs} = lists:foldl(DepthsToDeliver, {[], []}, SubsByDepth), - JIDSubs. +node_subscriptions(Host, Node, NodeId, Type, _NodeOptions, Notify) -> + % TODO temporary dirty condition, should be improved using plugin or node options + case Type of + ?STDNODE -> node_subscriptions_bare(Host, Node, NodeId, Type); + ?PEPNODE -> node_subscriptions_bare(Host, Node, NodeId, Type); + _ -> node_subscriptions_full(Host, Node, Notify) + end. + +node_subscriptions_bare(Host, Node, NodeId, Type) -> + case node_action(Host, Type, get_node_subscriptions, [NodeId]) of + {result, Subs} -> + SubsByJid = lists:foldl( + fun({JID, subscribed, SubId}, Acc) -> + case dict:is_key(JID, Acc) of + true -> dict:append(JID, SubId, Acc); + false -> dict:store(JID, [SubId], Acc) + end; + (_, Acc) -> + Acc + end, dict:new(), Subs), + [{J, Node, S} || {J, S} <- dict:to_list(SubsByJid)]; + _ -> + [] + end. + +node_subscriptions_full(Host, Node, NotifyType) -> + Action = fun() -> + Collection = tree_call(Host, get_parentnodes_tree, [Host, Node, service_jid(Host)]), + {result, [{Depth, [{N, sub_with_options(N)} || N <- Nodes]} || {Depth, Nodes} <- Collection]} + end, + case transaction(Host, Action, sync_dirty) of + {result, CollSubs} -> subscribed_nodes_by_jid(NotifyType, CollSubs); + _ -> [] + end. + +sub_with_options(#pubsub_node{type = Type, id = NodeId}) -> + case node_call(Type, get_node_subscriptions, [NodeId]) of + {result, Subs} -> + lists:foldl( + fun({JID, subscribed, SubId}, Acc) -> [sub_with_options(JID, NodeId, SubId) | Acc]; + (_, Acc) -> Acc + end, [], Subs); + _ -> + [] + end. +sub_with_options(JID, NodeId, SubId) -> + case pubsub_subscription_odbc:read_subscription(Entity, NodeId, SubId) of + {result, #pubsub_subscription{options = Options}} -> {JID, SubId, Options}; + _ -> {JID, SubId, []} + end. user_resources(User, Server) -> ejabberd_sm:get_user_resources(User, Server). +user_resource(User, Server, []) -> + case user_resources(User, Server) of + [R|_] -> R; + _ -> [] + end; +user_resource(_, _, Resource) -> + Resource. + %%%%%%% Configuration handling %%

There are several reasons why the default node configuration options request might fail:

@@ -4414,7 +4254,7 @@ extended_headers(JIDs) -> on_user_offline(_, JID, _) -> {User, Server, Resource} = jlib:short_prepd_jid(JID), - case ejabberd_sm:get_user_resources(User, Server) of + case user_resources(User, Server) of [] -> purge_offline({User, Server, Resource}); _ -> true end. diff --git a/src/mod_pubsub/pubsub_odbc.patch b/src/mod_pubsub/pubsub_odbc.patch index 2d22de654..a8910c86a 100644 --- a/src/mod_pubsub/pubsub_odbc.patch +++ b/src/mod_pubsub/pubsub_odbc.patch @@ -1,5 +1,5 @@ ---- mod_pubsub.erl 2010-12-07 19:19:44.000000000 +0100 -+++ mod_pubsub_odbc.erl 2010-12-07 19:30:26.000000000 +0100 +--- mod_pubsub.erl 2011-02-08 18:52:12.000000000 +0100 ++++ mod_pubsub_odbc.erl 2011-02-08 19:07:44.000000000 +0100 @@ -42,7 +42,7 @@ %%% 6.2.3.1, 6.2.3.5, and 6.3. For information on subscription leases see %%% XEP-0060 section 12.18. @@ -7,7 +7,7 @@ --module(mod_pubsub). +-module(mod_pubsub_odbc). -author('christophe.romain@process-one.net'). - -version('1.13-0'). + -version('1.13-1'). @@ -54,9 +54,9 @@ -include("pubsub.hrl"). @@ -22,7 +22,7 @@ %% exports for hooks -export([presence_probe/3, -@@ -103,7 +103,7 @@ +@@ -102,7 +102,7 @@ string_to_affiliation/1, extended_error/2, extended_error/3, @@ -31,7 +31,7 @@ ]). %% API and gen_server callbacks -@@ -122,7 +122,7 @@ +@@ -121,7 +121,7 @@ -export([send_loop/1 ]). @@ -40,7 +40,7 @@ -define(LOOPNAME, ejabberd_mod_pubsub_loop). -define(PLUGIN_PREFIX, "node_"). -define(TREE_PREFIX, "nodetree_"). -@@ -249,8 +249,6 @@ +@@ -248,8 +248,6 @@ ok end, ejabberd_router:register_route(Host), @@ -49,7 +49,7 @@ init_nodes(Host, ServerHost, NodeTree, Plugins), State = #state{host = Host, server_host = ServerHost, -@@ -352,229 +350,15 @@ +@@ -351,229 +349,15 @@ init_nodes(Host, ServerHost, _NodeTree, Plugins) -> %% TODO, this call should be done plugin side @@ -282,7 +282,7 @@ -spec(send_loop/1 :: ( State::#state{}) -@@ -591,7 +375,10 @@ +@@ -590,7 +374,10 @@ %% for each node From is subscribed to %% and if the node is so configured, send the last published item to From lists:foreach(fun(PType) -> @@ -294,7 +294,7 @@ lists:foreach( fun({Node, subscribed, _, SubJID}) -> if (SubJID == LJID) or (SubJID == BJID) -> -@@ -764,7 +551,8 @@ +@@ -763,7 +550,8 @@ [#xmlel{name = 'identity', ns = ?NS_DISCO_INFO, attrs = [?XMLATTR(<<"category">>, <<"pubsub">>), ?XMLATTR(<<"type">>, <<"pep">>)]}]; disco_identity(#jid{node = U, domain = S, resource = R} = Host, NodeId, From) -> @@ -304,7 +304,7 @@ case get_allowed_items_call(Host, NodeIdx, From, Type, Options, Owners) of {result, _} -> {result, -@@ -814,7 +602,8 @@ +@@ -813,7 +601,8 @@ [?NS_PUBSUB_s | [?NS_PUBSUB_s++"#"++Feature || Feature <- features("pep")]]; disco_features(#jid{node = U, domain = S, resource = R} = Host, NodeId, From) -> @@ -314,7 +314,7 @@ case get_allowed_items_call(Host, NodeIdx, From, Type, Options, Owners) of {result, _} -> {result, [?NS_PUBSUB_s -@@ -854,7 +643,8 @@ +@@ -853,7 +642,8 @@ ). disco_items(#jid{raw = JID, node = U, domain = S, resource = R} = Host, <<>>, From) -> @@ -324,7 +324,7 @@ case get_allowed_items_call(Host, NodeIdx, From, Type, Options, Owners) of {result, _} -> [#xmlel{name = 'item', ns = ?NS_DISCO_INFO, -@@ -868,13 +658,14 @@ +@@ -867,13 +657,14 @@ _ -> Acc end end, @@ -341,7 +341,7 @@ case get_allowed_items_call(Host, NodeIdx, From, Type, Options, Owners) of {result, Items} -> {result, -@@ -983,10 +774,10 @@ +@@ -982,10 +773,10 @@ lists:foreach(fun(PType) -> {result, Subscriptions} = node_action(Host, PType, get_entity_subscriptions, [Host, Entity]), lists:foreach(fun @@ -354,7 +354,7 @@ true -> node_action(Host, PType, unsubscribe_node, [Nidx, Entity, JID, all]); false -> -@@ -1184,9 +975,10 @@ +@@ -1183,9 +974,10 @@ end, ejabberd_router:route(To, From, Res); %% Service discovery : disco#items @@ -367,7 +367,7 @@ {result, IQRes} -> Result = #xmlel{ns = ?NS_DISCO_ITEMS, name = 'query', -@@ -1336,7 +1128,7 @@ +@@ -1335,7 +1127,7 @@ Types = case tree_call(Host, get_subnodes, [Host, NodeId, From]) of [] -> ["leaf"]; _ -> @@ -376,7 +376,7 @@ {result, []} -> ["collection"]; {result, _} -> ["leaf", "collection"]; _ -> [] -@@ -1354,10 +1146,15 @@ +@@ -1353,10 +1145,15 @@ [#xmlel{ns = ?NS_DISCO_INFO, name = 'feature', attrs = [?XMLATTR(<<"var">>, ?NS_PUBSUB_b)]} | @@ -396,7 +396,7 @@ end, case transaction(Host, NodeId, Action, sync_dirty) of {result, {_, Result}} -> {result, Result}; -@@ -1401,10 +1198,10 @@ +@@ -1400,10 +1197,10 @@ name = 'feature', attrs = [?XMLATTR(<<"var">>, ?NS_VCARD_b)]}] ++ @@ -411,7 +411,7 @@ iq_disco_info(Host, NodeId, From, _Lang) when NodeId == ?NS_ADHOC_b orelse NodeId == ?NS_PUBSUB_GET_PENDING_b -> command_disco_info(Host, NodeId, From); -@@ -1412,16 +1209,17 @@ +@@ -1411,16 +1208,17 @@ node_disco_info(Host, NodeId, From). @@ -432,7 +432,7 @@ case tree_action(Host, get_subnodes, [Host, <<>>, From]) of Nodes when is_list(Nodes) -> {result, lists:map( -@@ -1438,7 +1236,7 @@ +@@ -1437,7 +1235,7 @@ Other -> Other end; @@ -441,7 +441,7 @@ %% TODO: support localization of this string {result, [#xmlel{ns = ?NS_DISCO_ITEMS, -@@ -1446,14 +1244,15 @@ +@@ -1445,14 +1243,15 @@ attrs = [?XMLATTR(<<"jid">>, Host), ?XMLATTR(<<"node">>, ?NS_PUBSUB_GET_PENDING_b), ?XMLATTR(<<"name">>, "Get Pending")]}]}; @@ -462,7 +462,7 @@ end, Nodes = lists:map( fun(#pubsub_node{id = {_, SubNodeId}, options = SubOptions}) -> -@@ -1476,7 +1275,7 @@ +@@ -1475,7 +1274,7 @@ attrs = [?XMLATTR(<<"jid">>, Host), ?XMLATTR(<<"name">>, Name)]} end, NodeItems), @@ -471,7 +471,7 @@ end, case transaction(Host, NodeId, Action, sync_dirty) of {result, {_, Result}} -> {result, Result}; -@@ -1485,12 +1284,6 @@ +@@ -1484,12 +1283,6 @@ @@ -484,7 +484,7 @@ get_presence_and_roster_permissions(Host, From, Owners, AccessModel, AllowedGroups) -> if (AccessModel == presence) or (AccessModel == roster) -> case Host of -@@ -1646,7 +1439,7 @@ +@@ -1645,7 +1438,7 @@ end; (_, Acc) -> Acc end, [], exmpp_xml:remove_cdata_from_list(SubEls)), @@ -493,7 +493,7 @@ {'get', 'subscriptions'} -> get_subscriptions(Host, NodeId, From, Plugins); {'get', 'affiliations'} -> -@@ -1837,7 +1630,8 @@ +@@ -1836,7 +1629,8 @@ _ -> [] end end, @@ -503,7 +503,7 @@ sync_dirty) of {result, Res} -> Res; Err -> Err -@@ -1881,7 +1675,7 @@ +@@ -1880,7 +1674,7 @@ %%% authorization handling @@ -512,7 +512,7 @@ Lang = <<"en">>, %% TODO fix {U, S, R} = Subscriber, Stanza = #xmlel{ns = ?NS_JABBER_CLIENT, name = 'message', children = -@@ -1911,7 +1705,7 @@ +@@ -1910,7 +1704,7 @@ lists:foreach(fun(Owner) -> {U, S, R} = Owner, ejabberd_router:route(service_jid(Host), exmpp_jid:make(U, S, R), Stanza) @@ -521,7 +521,7 @@ find_authorization_response(Packet) -> Els = Packet#xmlel.children, -@@ -1970,8 +1764,8 @@ +@@ -1969,8 +1763,8 @@ "true" -> true; _ -> false end, @@ -532,16 +532,16 @@ {result, Subscriptions} = node_call(Type, get_subscriptions, [Nidx, Subscriber]), if not IsApprover -> -@@ -2170,7 +1964,7 @@ +@@ -2167,7 +1961,7 @@ end, Reply = #xmlel{ns = ?NS_PUBSUB, name = 'pubsub', children = [#xmlel{ns = ?NS_PUBSUB, name = 'create', attrs = nodeAttr(Node)}]}, - case transaction(CreateNode, transaction) of + case transaction(Host, CreateNode, transaction) of - {result, {NodeId, SubsByDepth, {Result, broadcast}}} -> - broadcast_created_node(Host, Node, NodeId, Type, NodeOptions, SubsByDepth), + {result, {NodeId, {Result, broadcast}}} -> + broadcast_created_node(Host, Node, NodeId, Type, NodeOptions), case Result of -@@ -2274,7 +2068,7 @@ +@@ -2269,7 +2063,7 @@ %%
  • The node does not exist.
    • %% subscribe_node(Host, Node, From, JID, Configuration) -> @@ -550,7 +550,7 @@ {result, GoodSubOpts} -> GoodSubOpts; _ -> invalid end, -@@ -2284,7 +2078,7 @@ +@@ -2279,7 +2073,7 @@ _:_ -> {undefined, undefined, undefined} end, @@ -559,7 +559,7 @@ Features = features(Type), SubscribeFeature = lists:member("subscribe", Features), OptionsFeature = lists:member("subscription-options", Features), -@@ -2293,6 +2087,7 @@ +@@ -2288,6 +2082,7 @@ AccessModel = get_option(Options, access_model), SendLast = get_option(Options, send_last_published_item), AllowedGroups = get_option(Options, roster_groups_allowed, []), @@ -567,7 +567,7 @@ {PresenceSubscription, RosterGroup} = get_presence_and_roster_permissions(Host, Subscriber, Owners, AccessModel, AllowedGroups), if not SubscribeFeature -> -@@ -2639,7 +2434,7 @@ +@@ -2633,7 +2428,7 @@ %%

    The permission are not checked in this function.

    %% @todo We probably need to check that the user doing the query has the right %% to read the items. @@ -576,7 +576,7 @@ MaxItems = if SMaxItems == "" -> get_max_items_node(Host); -@@ -2653,12 +2448,13 @@ +@@ -2647,12 +2442,13 @@ {error, Error} -> {error, Error}; _ -> @@ -591,7 +591,7 @@ {PresenceSubscription, RosterGroup} = get_presence_and_roster_permissions(Host, From, Owners, AccessModel, AllowedGroups), if not RetreiveFeature -> -@@ -2671,11 +2467,11 @@ +@@ -2665,11 +2461,11 @@ node_call(Type, get_items, [Nidx, From, AccessModel, PresenceSubscription, RosterGroup, @@ -605,7 +605,7 @@ SendItems = case ItemIds of [] -> Items; -@@ -2688,7 +2484,7 @@ +@@ -2682,7 +2478,7 @@ %% number of items sent to MaxItems: {result, #xmlel{ns = ?NS_PUBSUB, name = 'pubsub', children = [#xmlel{ns = ?NS_PUBSUB, name = 'items', attrs = nodeAttr(Node), children = @@ -614,7 +614,7 @@ Error -> Error end -@@ -2729,6 +2525,17 @@ +@@ -2723,6 +2519,17 @@ {result, {_, Items}} -> Items; Error -> Error end. @@ -632,7 +632,7 @@ %% @spec (Host, Node, NodeId, Type, LJID, Number) -> any() %% Host = pubsubHost() -@@ -2740,16 +2547,29 @@ +@@ -2734,16 +2541,29 @@ %% @doc

    Resend the items of a node to the user.

    %% @todo use cache-last-item feature send_items(Host, Node, NodeId, Type, LJID, 'last') -> @@ -668,7 +668,7 @@ send_items(Host, Node, NodeId, Type, {LU, LS, LR} = LJID, Number) -> ToSend = case node_action(Host, Type, get_items, [NodeId, LJID]) of {result, []} -> -@@ -2876,7 +2696,8 @@ +@@ -2870,7 +2690,8 @@ error -> {error, 'bad-request'}; _ -> @@ -678,7 +678,7 @@ case lists:member(Owner, Owners) of true -> OwnerJID = exmpp_jid:make(Owner), -@@ -2886,24 +2707,8 @@ +@@ -2880,24 +2701,8 @@ end, lists:foreach( fun({JID, Affiliation}) -> @@ -705,7 +705,7 @@ end, FilteredEntities), {result, []}; _ -> -@@ -2958,11 +2763,11 @@ +@@ -2952,11 +2757,11 @@ end. read_sub(Subscriber, Node, NodeId, SubId, Lang) -> @@ -719,7 +719,7 @@ OptionsEl = #xmlel{ns = ?NS_PUBSUB, name = 'options', attrs = [ ?XMLATTR(<<"jid">>, exmpp_jid:to_binary(Subscriber)), ?XMLATTR(<<"subid">>, SubId) | nodeAttr(Node)], -@@ -2989,7 +2794,7 @@ +@@ -2983,7 +2788,7 @@ end. set_options_helper(Configuration, JID, NodeId, SubId, Type) -> @@ -728,7 +728,7 @@ {result, GoodSubOpts} -> GoodSubOpts; _ -> invalid end, -@@ -3019,7 +2824,7 @@ +@@ -3013,7 +2818,7 @@ write_sub(_Subscriber, _NodeId, _SubId, invalid) -> {error, extended_error('bad-request', "invalid-options")}; write_sub(Subscriber, NodeId, SubId, Options) -> @@ -737,7 +737,7 @@ {error, notfound} -> {error, extended_error('not-acceptable', "invalid-subid")}; {result, _} -> -@@ -3193,8 +2998,8 @@ +@@ -3187,8 +2992,8 @@ ?XMLATTR(<<"subsription">>, subscription_to_string(Sub)) | nodeAttr(Node)]}]}]}, ejabberd_router:route(service_jid(Host), JID, Stanza) end, @@ -748,25 +748,27 @@ true -> Result = lists:foldl(fun({JID, Subscription, SubId}, Acc) -> -@@ -3557,7 +3362,7 @@ - {Depth, [{Node, get_node_subs(Node)} || Node <- Nodes]} - end, tree_call(Host, get_parentnodes_tree, [Host, NodeId, service_jid(Host)]))} - end, -- case transaction(Action, sync_dirty) of -+ case transaction(Host, Action, sync_dirty) of - {result, CollSubs} -> CollSubs; +@@ -3542,7 +3347,7 @@ + Collection = tree_call(Host, get_parentnodes_tree, [Host, Node, service_jid(Host)]), + {result, [{Depth, [{N, sub_with_options(N)} || N <- Nodes]} || {Depth, Nodes} <- Collection]} + end, +- case transaction(Action, sync_dirty) of ++ case transaction(Host, Action, sync_dirty) of + {result, CollSubs} -> subscribed_nodes_by_jid(NotifyType, CollSubs); _ -> [] + end. +@@ -3558,8 +3363,8 @@ + [] + end. + sub_with_options(JID, NodeId, SubId) -> +- case pubsub_subscription:read_subscription(JID, NodeId, SubId) of +- #pubsub_subscription{options = Options} -> {JID, SubId, Options}; ++ case pubsub_subscription_odbc:read_subscription(Entity, NodeId, SubId) of ++ {result, #pubsub_subscription{options = Options}} -> {JID, SubId, Options}; + _ -> {JID, SubId, []} end. -@@ -3587,7 +3392,7 @@ - get_options_for_subs(NodeIdx, Subs) -> - lists:foldl(fun({Entity, 'subscribed', SubId}, Acc) -> -- case pubsub_subscription:read_subscription(Entity, NodeIdx, SubId) of -+ case pubsub_subscription_odbc:read_subscription(Entity, NodeIdx, SubId) of - {error, 'notfound'} -> [{Entity, SubId, []} | Acc]; - #pubsub_subscription{options = Options} -> [{Entity, SubId, Options} | Acc] - end; -@@ -3831,6 +3636,30 @@ +@@ -3671,6 +3476,30 @@ Result end. @@ -797,7 +799,7 @@ %% @spec (Host, Options) -> MaxItems %% Host = host() %% Options = [Option] -@@ -4354,9 +4183,14 @@ +@@ -4194,9 +4023,14 @@ tree_action(Host, Function, Args) -> ?DEBUG("tree_action ~p ~p ~p",[Host,Function,Args]), @@ -815,7 +817,7 @@ %% @doc

    node plugin call.

    -spec(node_call/3 :: -@@ -4394,7 +4228,7 @@ +@@ -4234,7 +4068,7 @@ node_action(Host, Type, Function, Args) -> ?DEBUG("node_action ~p ~p ~p ~p",[Host,Type,Function,Args]), @@ -824,7 +826,7 @@ node_call(Type, Function, Args) end, sync_dirty). -@@ -4409,7 +4243,7 @@ +@@ -4249,7 +4083,7 @@ ). transaction(Host, NodeId, Action, Trans) -> @@ -833,7 +835,7 @@ case tree_call(Host, get_node, [Host, NodeId]) of #pubsub_node{} = Node -> case Action(Node) of -@@ -4423,7 +4257,7 @@ +@@ -4263,7 +4097,7 @@ end, Trans). @@ -842,7 +844,7 @@ ( Host :: string() | host(), Action :: fun(), -@@ -4431,21 +4265,28 @@ +@@ -4271,21 +4105,28 @@ -> {'result', Nodes :: [] | [Node::pubsubNode()]} ). @@ -876,7 +878,7 @@ {result, Result} -> {result, Result}; {error, Error} -> {error, Error}; {atomic, {result, Result}} -> {result, Result}; -@@ -4453,6 +4294,15 @@ +@@ -4293,6 +4134,15 @@ {aborted, Reason} -> ?ERROR_MSG("transaction return internal error: ~p~n", [{aborted, Reason}]), {error, 'internal-server-error'}; @@ -892,7 +894,7 @@ {'EXIT', Reason} -> ?ERROR_MSG("transaction return internal error: ~p~n", [{'EXIT', Reason}]), {error, 'internal-server-error'}; -@@ -4461,6 +4311,16 @@ +@@ -4301,6 +4151,16 @@ {error, 'internal-server-error'} end.