Chapril
/
xmpp.chapril.org-ejabberd
mirror of https://github.com/processone/ejabberd.git
24
1
Fork 0
Code Releases Activity

refactor pubsub broadcast to allow large optimization

This commit is contained in:
Christophe Romain 2011-02-08 19:08:38 +01:00
parent a3dba06baf
commit 2f681b05f3
4 changed files with 420 additions and 726 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

12
src/ejabberd_sm.erl
View File

@ -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},

494
src/mod_pubsub/mod_pubsub.erl
View File

@ -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 <p>Build pubsub event stanza</p>
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
%%<p>There are several reasons why the default node configuration options request might fail:</p>
@ -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.

502
src/mod_pubsub/mod_pubsub_odbc.erl
View File

@ -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 <p>Build pubsub event stanza</p>
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
%%<p>There are several reasons why the default node configuration options request might fail:</p>
@ -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.

138
src/mod_pubsub/pubsub_odbc.patch
View File

@ -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 @@
%%<li>The node does not exist.</li>
%%</ul>
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 @@
%% <p>The permission are not checked in this function.</p>
%% @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 <p>Resend the items of a node to the user.</p>
%% @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 <p>node plugin call.</p>
-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.