Chapril/xmpp.chapril.org-ejabberd
25
1
Fork 0
mirror of https://github.com/processone/ejabberd.git synced 2024-12-26 17:38:45 +01:00
Code Releases Activity

port previous pubsub changes to _odbc

Browse Source
This commit is contained in:
Christophe Romain 2011-02-08 18:29:52 +01:00
parent f485109c39
commit 024a80d41f
2 changed files with 243 additions and 300 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

374
src/mod_pubsub/mod_pubsub_odbc.erl
View File

@ -78,6 +78,7 @@
%% exports for console debug manual use %% exports for console debug manual use
-export([create_node/5, -export([create_node/5,
create_node/7,
delete_node/3, delete_node/3,
subscribe_node/5, subscribe_node/5,
unsubscribe_node/5, unsubscribe_node/5,
@ -87,7 +88,6 @@
get_items/2, get_items/2,
get_item/3, get_item/3,
get_cached_item/2, get_cached_item/2,
broadcast_stanza/9,
get_configure/5, get_configure/5,
set_configure/5, set_configure/5,
tree_action/3, tree_action/3,
@ -1574,10 +1574,8 @@ create_node(Host, ServerHost, Node, Owner, GivenType, Access, Configuration) ->
{result, true} -> {result, true} ->
case tree_call(Host, create_node, [Host, Node, Type, Owner, NodeOptions, Parents]) of case tree_call(Host, create_node, [Host, Node, Type, Owner, NodeOptions, Parents]) of
{ok, NodeId} -> {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 case node_call(Type, create_node, [NodeId, Owner]) of
{result, Result} -> {result, {NodeId, SubsByDepth, Result}}; {result, Result} -> {result, {NodeId, Result}};
Error -> Error Error -> Error
end; end;
{error, {virtual, NodeId}} -> {error, {virtual, NodeId}} ->
@ -1596,15 +1594,15 @@ create_node(Host, ServerHost, Node, Owner, GivenType, Access, Configuration) ->
[{xmlelement, "create", nodeAttr(Node), [{xmlelement, "create", nodeAttr(Node),
[]}]}], []}]}],
case transaction(Host, CreateNode, transaction) of case transaction(Host, CreateNode, transaction) of
{result, {NodeId, SubsByDepth, {Result, broadcast}}} -> {result, {NodeId, {Result, broadcast}}} ->
broadcast_created_node(Host, Node, NodeId, Type, NodeOptions, SubsByDepth), broadcast_created_node(Host, Node, NodeId, Type, NodeOptions),
case Result of case Result of
default -> {result, Reply}; default -> {result, Reply};
_ -> {result, Result} _ -> {result, Result}
end; end;
{result, {_NodeId, _SubsByDepth, default}} -> {result, {_NodeId, default}} ->
{result, Reply}; {result, Reply};
{result, {_NodeId, _SubsByDepth, Result}} -> {result, {_NodeId, Result}} ->
{result, Result}; {result, Result};
Error -> Error ->
%% in case we change transaction to sync_dirty... %% in case we change transaction to sync_dirty...
@ -1636,11 +1634,9 @@ delete_node(Host, Node, Owner) ->
Action = fun(#pubsub_node{type = Type, id = NodeId}) -> Action = fun(#pubsub_node{type = Type, id = NodeId}) ->
case node_call(Type, get_affiliation, [NodeId, Owner]) of case node_call(Type, get_affiliation, [NodeId, Owner]) of
{result, owner} -> {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]), Removed = tree_call(Host, delete_node, [Host, Node]),
case node_call(Type, delete_node, [Removed]) of case node_call(Type, delete_node, [Removed]) of
{result, Res} -> {result, {SubsByDepth, Res}}; {result, Res} -> {result, Res};
Error -> Error Error -> Error
end; end;
_ -> _ ->
@ -1650,26 +1646,26 @@ delete_node(Host, Node, Owner) ->
end, end,
Reply = [], Reply = [],
case transaction(Host, Node, Action, transaction) of case transaction(Host, Node, Action, transaction) of
{result, {_, {SubsByDepth, {Result, broadcast, Removed}}}} -> {result, {_, {Result, broadcast, Removed}}} ->
lists:foreach(fun({RNode, _RSubscriptions}) -> lists:foreach(fun({RNode, _RSubscriptions}) ->
{RH, RN} = RNode#pubsub_node.nodeid, {RH, RN} = RNode#pubsub_node.nodeid,
NodeId = RNode#pubsub_node.id, NodeId = RNode#pubsub_node.id,
Type = RNode#pubsub_node.type, Type = RNode#pubsub_node.type,
Options = RNode#pubsub_node.options, Options = RNode#pubsub_node.options,
broadcast_removed_node(RH, RN, NodeId, Type, Options, SubsByDepth) broadcast_removed_node(RH, RN, NodeId, Type, Options)
end, Removed), end, Removed),
case Result of case Result of
default -> {result, Reply}; default -> {result, Reply};
_ -> {result, Result} _ -> {result, Result}
end; end;
{result, {_, {_, {Result, _Removed}}}} -> {result, {_, {Result, _Removed}}} ->
case Result of case Result of
default -> {result, Reply}; default -> {result, Reply};
_ -> {result, Result} _ -> {result, Result}
end; end;
{result, {_, {_, default}}} -> {result, {_, default}} ->
{result, Reply}; {result, Reply};
{result, {_, {_, Result}}} -> {result, {_, Result}} ->
{result, Result}; {result, Result};
Error -> Error ->
Error Error
@ -1877,12 +1873,11 @@ publish_item(Host, ServerHost, Node, Publisher, ItemId, Payload) ->
NodeId = TNode#pubsub_node.id, NodeId = TNode#pubsub_node.id,
Type = TNode#pubsub_node.type, Type = TNode#pubsub_node.type,
Options = TNode#pubsub_node.options, Options = TNode#pubsub_node.options,
BroadcastPayload = case Broadcast of BrPayload = case Broadcast of
default -> Payload;
broadcast -> Payload; broadcast -> Payload;
PluginPayload -> PluginPayload PluginPayload -> PluginPayload
end, end,
broadcast_publish_item(Host, Node, NodeId, Type, Options, Removed, ItemId, jlib:jid_tolower(Publisher), BroadcastPayload), broadcast_publish_item(Host, Node, NodeId, Type, Options, ItemId, jlib:jid_tolower(Publisher), BrPayload, Removed),
set_cached_item(Host, NodeId, ItemId, Publisher, Payload), set_cached_item(Host, NodeId, ItemId, Publisher, Payload),
case Result of case Result of
default -> {result, Reply}; default -> {result, Reply};
@ -2708,6 +2703,9 @@ sub_to_deliver(_LJID, NotifyType, Depth, SubOptions) ->
sub_option_can_deliver(NotifyType, Depth, Option) sub_option_can_deliver(NotifyType, Depth, Option)
end, SubOptions). 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(items, _, {subscription_type, nodes}) -> false;
sub_option_can_deliver(nodes, _, {subscription_type, items}) -> false; sub_option_can_deliver(nodes, _, {subscription_type, items}) -> false;
sub_option_can_deliver(_, _, {subscription_depth, all}) -> true; sub_option_can_deliver(_, _, {subscription_depth, all}) -> true;
@ -2716,13 +2714,9 @@ sub_option_can_deliver(_, _, {deliver, false}) -> false;
sub_option_can_deliver(_, _, {expire, When}) -> now() < When; sub_option_can_deliver(_, _, {expire, When}) -> now() < When;
sub_option_can_deliver(_, _, _) -> true. sub_option_can_deliver(_, _, _) -> true.
node_to_deliver(LJID, NodeOptions) ->
PresenceDelivery = get_option(NodeOptions, presence_based_delivery),
presence_can_deliver(LJID, PresenceDelivery).
presence_can_deliver(_, false) -> true; presence_can_deliver(_, false) -> true;
presence_can_deliver({User, Server, Resource}, 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; [] -> false;
Ss -> Ss ->
lists:foldl(fun(_, true) -> true; lists:foldl(fun(_, true) -> true;
@ -2804,223 +2798,149 @@ event_stanza_withmoreels(Els, MoreEls) ->
{xmlelement, "message", [], {xmlelement, "message", [],
[{xmlelement, "event", [{"xmlns", ?NS_PUBSUB_EVENT}], Els} | MoreEls]}. [{xmlelement, "event", [{"xmlns", ?NS_PUBSUB_EVENT}], Els} | MoreEls]}.
event_stanza(Event, EvAttr) ->
event_stanza([{xmlelement, Event, EvAttr, []}]).
event_stanza(Event, EvAttr, Entries) ->
event_stanza([{xmlelement, Event, EvAttr,
[{xmlelement, Entry, EnAttr, []} || {Entry, EnAttr} <- Entries]}]).
event_stanza(Event, EvAttr, Entry, EnAttr, Payload) ->
event_stanza([{xmlelement, Event, EvAttr, [{xmlelement, Entry, EnAttr, 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 functions
broadcast_publish_item(Host, Node, NodeId, Type, NodeOptions, Removed, ItemId, From, Payload) -> broadcast_publish_item(Host, Node, NodeId, Type, NodeOptions, ItemId, Publisher, Payload, Removed) ->
case get_collection_subscriptions(Host, Node) of Publish = case get_option(NodeOptions, deliver_payloads) of
SubsByDepth when is_list(SubsByDepth) -> true -> event_stanza("items", nodeAttr(Node), "item", itemAttr(ItemId), Payload, Publisher);
Content = case get_option(NodeOptions, deliver_payloads) of false -> event_stanza("items", nodeAttr(Node), "item", itemAttr(ItemId), [], Publisher)
true -> Payload;
false -> []
end, end,
Stanza = event_stanza(
[{xmlelement, "items", nodeAttr(Node),
[{xmlelement, "item", itemAttr(ItemId), Content}]}]),
broadcast_stanza(Host, From, Node, NodeId, Type,
NodeOptions, SubsByDepth, items, Stanza, true),
case Removed of case Removed of
[] -> [] ->
ok; {result, broadcast(Host, Node, NodeId, Type, NodeOptions, items, true, Publish, true)};
_ -> _ ->
case get_option(NodeOptions, notify_retract) of Retract = event_stanza("items", nodeAttr(Node), [{"retract", itemAttr(Rid)} || Rid <- Removed]),
true -> Stanzas = [{true, Publish, true}, {get_option(NodeOptions, notify_retract), Retract, true}],
RetractStanza = event_stanza( {result, broadcast(Host, Node, NodeId, Type, NodeOptions, items, Stanzas)}
[{xmlelement, "items", nodeAttr(Node),
[{xmlelement, "retract", itemAttr(RId), []} || RId <- Removed]}]),
broadcast_stanza(Host, Node, NodeId, Type,
NodeOptions, SubsByDepth,
items, RetractStanza, true);
_ ->
ok
end
end,
{result, true};
_ ->
{result, false}
end. end.
broadcast_retract_items(Host, Node, NodeId, Type, NodeOptions, ItemIds) -> 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, ItemIds, notify_retract).
broadcast_retract_items(_Host, _Node, _NodeId, _Type, _NodeOptions, [], _ForceNotify) -> broadcast_retract_items(_Host, _Node, _NodeId, _Type, _NodeOptions, [], _) ->
{result, false}; {result, false};
broadcast_retract_items(Host, Node, NodeId, Type, NodeOptions, ItemIds, ForceNotify) -> broadcast_retract_items(Host, Node, NodeId, Type, NodeOptions, ItemIds, false) ->
case (get_option(NodeOptions, notify_retract) or ForceNotify) of broadcast_retract_items(Host, Node, NodeId, Type, NodeOptions, ItemIds, notify_retract);
true -> broadcast_retract_items(Host, Node, NodeId, Type, NodeOptions, ItemIds, Notify) ->
case get_collection_subscriptions(Host, Node) of Stanza = event_stanza("items", nodeAttr(Node), [{"retract", itemAttr(Rid)} || Rid <- ItemIds]),
SubsByDepth when is_list(SubsByDepth) -> {result, broadcast(Host, Node, NodeId, Type, NodeOptions, items, Notify, Stanza, true)}.
Stanza = event_stanza(
[{xmlelement, "items", nodeAttr(Node),
[{xmlelement, "retract", itemAttr(ItemId), []} || ItemId <- ItemIds]}]),
broadcast_stanza(Host, Node, NodeId, Type,
NodeOptions, SubsByDepth, items, Stanza, true),
{result, true};
_ ->
{result, false}
end;
_ ->
{result, false}
end.
broadcast_purge_node(Host, Node, NodeId, Type, NodeOptions) -> broadcast_purge_node(Host, Node, NodeId, Type, NodeOptions) ->
case get_option(NodeOptions, notify_retract) of Stanza = event_stanza("purge", nodeAttr(Node)),
true -> {result, broadcast(Host, Node, NodeId, Type, NodeOptions, nodes, notify_retract, Stanza, false)}.
case get_collection_subscriptions(Host, Node) of
SubsByDepth when is_list(SubsByDepth) ->
Stanza = event_stanza(
[{xmlelement, "purge", nodeAttr(Node),
[]}]),
broadcast_stanza(Host, Node, NodeId, Type,
NodeOptions, SubsByDepth, nodes, Stanza, false),
{result, true};
_ ->
{result, false}
end;
_ ->
{result, false}
end.
broadcast_removed_node(Host, Node, NodeId, Type, NodeOptions, SubsByDepth) -> broadcast_removed_node(Host, Node, NodeId, Type, NodeOptions) ->
case get_option(NodeOptions, notify_delete) of Stanza = event_stanza("delete", nodeAttr(Node)),
true -> {result, broadcast(Host, Node, NodeId, Type, NodeOptions, nodes, notify_delete, Stanza, false)}.
case SubsByDepth of
[] ->
{result, false};
_ ->
Stanza = event_stanza(
[{xmlelement, "delete", nodeAttr(Node),
[]}]),
broadcast_stanza(Host, Node, NodeId, Type,
NodeOptions, SubsByDepth, nodes, Stanza, false),
{result, true}
end;
_ ->
{result, false}
end.
broadcast_created_node(_, _, _, _, _, []) -> broadcast_created_node(Host, Node, NodeId, Type, NodeOptions) ->
{result, false}; Stanza = event_stanza("create", nodeAttr(Node)),
broadcast_created_node(Host, Node, NodeId, Type, NodeOptions, SubsByDepth) -> {result, broadcast(Host, Node, NodeId, Type, NodeOptions, nodes, true, Stanza, true)}.
Stanza = event_stanza([{xmlelement, "create", nodeAttr(Node), []}]),
broadcast_stanza(Host, Node, NodeId, Type, NodeOptions, SubsByDepth, nodes, Stanza, true),
{result, true}.
broadcast_config_notification(Host, Node, NodeId, Type, NodeOptions, Lang) -> broadcast_config_notification(Host, Node, NodeId, Type, NodeOptions, Lang) ->
case get_option(NodeOptions, notify_config) of Stanza = case get_option(NodeOptions, deliver_payloads) of
true -> true ->
case get_collection_subscriptions(Host, Node) of event_stanza("configuration", nodeAttr(Node),
SubsByDepth when is_list(SubsByDepth) -> "x", [{"xmlns", ?NS_XDATA}, {"type", "result"}],
Content = case get_option(NodeOptions, deliver_payloads) of get_configure_xfields(Type, NodeOptions, Lang, []));
true ->
[{xmlelement, "x", [{"xmlns", ?NS_XDATA}, {"type", "result"}],
get_configure_xfields(Type, NodeOptions, Lang, [])}];
false -> false ->
[] event_stanza("configuration", nodeAttr(Node))
end, end,
Stanza = event_stanza( {result, broadcast(Host, Node, NodeId, Type, NodeOptions, nodes, notify_config, Stanza, false)}.
[{xmlelement, "configuration", nodeAttr(Node), Content}]),
broadcast_stanza(Host, Node, NodeId, Type,
NodeOptions, SubsByDepth, nodes, Stanza, false),
{result, true};
_ ->
{result, false}
end;
_ ->
{result, false}
end.
get_collection_subscriptions(Host, Node) -> broadcast(Host, Node, NodeId, Type, NodeOptions, Notify, Stanzas) ->
Action = fun() -> Subs = node_subscriptions(Host, Node, NodeId, Type, NodeOptions, Notify),
{result, lists:map(fun({Depth, Nodes}) -> Result = [broadcast(Host, Node, NodeId, Type, NodeOptions, Subs, Stanza, SHIM) ||
{Depth, [{N, get_node_subs(N)} || N <- Nodes]} {Cond, Stanza, SHIM} <- Stanzas, Cond =:= true],
end, tree_call(Host, get_parentnodes_tree, [Host, Node, service_jid(Host)]))} lists:member(true, Result).
end, broadcast(Host, Node, NodeId, Type, NodeOptions, Notify, true, Stanza, SHIM) ->
case transaction(Host, Action, sync_dirty) of Subs = node_subscriptions(Host, Node, NodeId, Type, NodeOptions, Notify),
{result, CollSubs} -> CollSubs; broadcast(Host, Node, NodeId, Type, NodeOptions, Subs, Stanza, SHIM);
_ -> [] broadcast(_Host, _Node, _NodeId, _Type, _NodeOptions, _Notify, false, _Stanza, _SHIM) ->
end. false;
broadcast(Host, Node, NodeId, Type, NodeOptions, Notify, Condition, Stanza, SHIM) ->
broadcast(Host, Node, NodeId, Type, NodeOptions, Notify, get_option(NodeOptions, Condition), Stanza, SHIM).
get_node_subs(#pubsub_node{type = Type, broadcast({U, S, R}, Node, NodeId, Type, NodeOptions, Subscriptions, Stanza, SHIM) ->
id = NodeID}) -> broadcast(S, Node, NodeId, Type, NodeOptions, Subscriptions, Stanza, SHIM)
case node_call(Type, get_node_subscriptions, [NodeID]) of or case ejabberd_sm:get_session_pid(U, S, user_resource(U, S, R)) of
{result, Subs} -> get_options_for_subs(NodeID, Subs);
Other -> Other
end.
get_options_for_subs(NodeID, Subs) ->
lists:foldl(fun({JID, subscribed, SubID}, Acc) ->
case pubsub_subscription_odbc:get_subscription(JID, NodeID, SubID) of
{error, notfound} -> [{JID, SubID, []} | Acc];
{result, #pubsub_subscription{options = Options}} -> [{JID, SubID, Options} | Acc];
_ -> Acc
end;
(_, Acc) ->
Acc
end, [], Subs).
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;
%% If there's only one SubID, don't add it
{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, jlib:make_jid(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),
%% Handles implicit presence subscriptions
SenderResource = case LResource of
[] ->
case user_resources(LUser, LServer) of
[Resource|_] -> Resource;
_ -> ""
end;
_ ->
LResource
end,
case ejabberd_sm:get_session_pid(LUser, LServer, SenderResource) of
C2SPid when is_pid(C2SPid) -> 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 %% 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 %% Also, add "replyto" if entity has presence subscription to the account owner
%% See XEP-0163 1.1 section 4.3.1 %% See XEP-0163 1.1 section 4.3.1
ejabberd_c2s:broadcast(C2SPid, Event = {pep_message, binary_to_list(Node)++"+notify"},
{pep_message, binary_to_list(Node)++"+notify"}, Message = case get_option(NodeOptions, notification_type, headline) of
_Sender = jlib:make_jid(LUser, LServer, ""), normal -> Stanza;
_StanzaToSend = add_extended_headers(Stanza, MsgType -> add_message_type(Stanza, atom_to_list(MsgType))
_ReplyTo = extended_headers([jlib:jid_to_string(Publisher)]))); 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; end;
broadcast_stanza(Host, _Publisher, Node, NodeId, Type, NodeOptions, SubsByDepth, NotifyType, BaseStanza, SHIM) -> broadcast(_Host, _Node, _NodeId, _Type, _NodeOptions, [], _Stanza, _SHIM) ->
broadcast_stanza(Host, Node, NodeId, Type, NodeOptions, SubsByDepth, NotifyType, BaseStanza, 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.
node_subscriptions(Host, Node, NodeId, Type, _NodeOptions, Notify) ->
% TODO temporary dirty condition, should be improved using plugin or node options
case Type of
"flat" -> node_subscriptions_bare(Host, Node, NodeId, Type);
"pep" -> 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.
subscribed_nodes_by_jid(NotifyType, SubsByDepth) -> subscribed_nodes_by_jid(NotifyType, SubsByDepth) ->
NodesToDeliver = fun(Depth, Node, Subs, Acc) -> NodesToDeliver = fun(Depth, Node, Subs, Acc) ->
@ -3075,9 +2995,33 @@ subscribed_nodes_by_jid(NotifyType, SubsByDepth) ->
{_, JIDSubs} = lists:foldl(DepthsToDeliver, {[], []}, SubsByDepth), {_, JIDSubs} = lists:foldl(DepthsToDeliver, {[], []}, SubsByDepth),
JIDSubs. JIDSubs.
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:get_subscription(JID, NodeId, SubId) of
{result, #pubsub_subscription{options = Options}} -> {JID, SubId, Options};
_ -> {JID, SubId, []}
end.
user_resources(User, Server) -> user_resources(User, Server) ->
ejabberd_sm:get_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 %%%%%%% Configuration handling
%%<p>There are several reasons why the default node configuration options request might fail:</p> %%<p>There are several reasons why the default node configuration options request might fail:</p>
@ -3736,7 +3680,7 @@ extended_headers(Jids) ->
on_user_offline(_, JID, _) -> on_user_offline(_, JID, _) ->
{User, Server, Resource} = jlib:jid_tolower(JID), {User, Server, Resource} = jlib:jid_tolower(JID),
case ejabberd_sm:get_user_resources(User, Server) of case user_resources(User, Server) of
[] -> purge_offline({User, Server, Resource}); [] -> purge_offline({User, Server, Resource});
_ -> true _ -> true
end. end.

81
src/mod_pubsub/pubsub_odbc.patch
View File

@ -1,5 +1,5 @@
--- mod_pubsub.erl 2010-12-07 13:54:26.000000000 +0100 --- mod_pubsub.erl 2011-02-08 16:50:27.000000000 +0100
+++ mod_pubsub_odbc.erl 2010-12-07 13:59:56.000000000 +0100 +++ mod_pubsub_odbc.erl 2011-02-08 18:27:16.000000000 +0100
@@ -42,7 +42,7 @@ @@ -42,7 +42,7 @@
%%% 6.2.3.1, 6.2.3.5, and 6.3. For information on subscription leases see %%% 6.2.3.1, 6.2.3.5, and 6.3. For information on subscription leases see
%%% XEP-0060 section 12.18. %%% XEP-0060 section 12.18.
@ -496,16 +496,16 @@
{result, Subscriptions} = node_call(Type, get_subscriptions, [NodeId, Subscriber]), {result, Subscriptions} = node_call(Type, get_subscriptions, [NodeId, Subscriber]),
if if
not IsApprover -> not IsApprover ->
@@ -1780,7 +1595,7 @@ @@ -1778,7 +1593,7 @@
Reply = [{xmlelement, "pubsub", [{"xmlns", ?NS_PUBSUB}], Reply = [{xmlelement, "pubsub", [{"xmlns", ?NS_PUBSUB}],
[{xmlelement, "create", nodeAttr(Node), [{xmlelement, "create", nodeAttr(Node),
[]}]}], []}]}],
- case transaction(CreateNode, transaction) of - case transaction(CreateNode, transaction) of
+ case transaction(Host, CreateNode, transaction) of + case transaction(Host, CreateNode, transaction) of
{result, {NodeId, SubsByDepth, {Result, broadcast}}} -> {result, {NodeId, {Result, broadcast}}} ->
broadcast_created_node(Host, Node, NodeId, Type, NodeOptions, SubsByDepth), broadcast_created_node(Host, Node, NodeId, Type, NodeOptions),
case Result of case Result of
@@ -1883,7 +1698,7 @@ @@ -1879,7 +1694,7 @@
%%<li>The node does not exist.</li> %%<li>The node does not exist.</li>
%%</ul> %%</ul>
subscribe_node(Host, Node, From, JID, Configuration) -> subscribe_node(Host, Node, From, JID, Configuration) ->
@ -514,7 +514,7 @@
{result, GoodSubOpts} -> GoodSubOpts; {result, GoodSubOpts} -> GoodSubOpts;
_ -> invalid _ -> invalid
end, end,
@@ -1891,7 +1706,7 @@ @@ -1887,7 +1702,7 @@
error -> {"", "", ""}; error -> {"", "", ""};
J -> jlib:jid_tolower(J) J -> jlib:jid_tolower(J)
end, end,
@ -523,7 +523,7 @@
Features = features(Type), Features = features(Type),
SubscribeFeature = lists:member("subscribe", Features), SubscribeFeature = lists:member("subscribe", Features),
OptionsFeature = lists:member("subscription-options", Features), OptionsFeature = lists:member("subscription-options", Features),
@@ -1900,6 +1715,7 @@ @@ -1896,6 +1711,7 @@
AccessModel = get_option(Options, access_model), AccessModel = get_option(Options, access_model),
SendLast = get_option(Options, send_last_published_item), SendLast = get_option(Options, send_last_published_item),
AllowedGroups = get_option(Options, roster_groups_allowed, []), AllowedGroups = get_option(Options, roster_groups_allowed, []),
@ -531,7 +531,7 @@
{PresenceSubscription, RosterGroup} = get_presence_and_roster_permissions(Host, Subscriber, Owners, AccessModel, AllowedGroups), {PresenceSubscription, RosterGroup} = get_presence_and_roster_permissions(Host, Subscriber, Owners, AccessModel, AllowedGroups),
if if
not SubscribeFeature -> not SubscribeFeature ->
@@ -2231,7 +2047,7 @@ @@ -2226,7 +2042,7 @@
%% <p>The permission are not checked in this function.</p> %% <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 %% @todo We probably need to check that the user doing the query has the right
%% to read the items. %% to read the items.
@ -540,7 +540,7 @@
MaxItems = MaxItems =
if if
SMaxItems == "" -> get_max_items_node(Host); SMaxItems == "" -> get_max_items_node(Host);
@@ -2245,12 +2061,13 @@ @@ -2240,12 +2056,13 @@
{error, Error} -> {error, Error} ->
{error, Error}; {error, Error};
_ -> _ ->
@ -555,7 +555,7 @@
{PresenceSubscription, RosterGroup} = get_presence_and_roster_permissions(Host, From, Owners, AccessModel, AllowedGroups), {PresenceSubscription, RosterGroup} = get_presence_and_roster_permissions(Host, From, Owners, AccessModel, AllowedGroups),
if if
not RetreiveFeature -> not RetreiveFeature ->
@@ -2263,11 +2080,11 @@ @@ -2258,11 +2075,11 @@
node_call(Type, get_items, node_call(Type, get_items,
[NodeId, From, [NodeId, From,
AccessModel, PresenceSubscription, RosterGroup, AccessModel, PresenceSubscription, RosterGroup,
@ -569,7 +569,7 @@
SendItems = case ItemIDs of SendItems = case ItemIDs of
[] -> [] ->
Items; Items;
@@ -2280,7 +2097,8 @@ @@ -2275,7 +2092,8 @@
%% number of items sent to MaxItems: %% number of items sent to MaxItems:
{result, [{xmlelement, "pubsub", [{"xmlns", ?NS_PUBSUB}], {result, [{xmlelement, "pubsub", [{"xmlns", ?NS_PUBSUB}],
[{xmlelement, "items", nodeAttr(Node), [{xmlelement, "items", nodeAttr(Node),
@ -579,7 +579,7 @@
Error -> Error ->
Error Error
end end
@@ -2302,10 +2120,15 @@ @@ -2297,10 +2115,15 @@
Error -> Error Error -> Error
end. end.
get_allowed_items_call(Host, NodeIdx, From, Type, Options, Owners) -> get_allowed_items_call(Host, NodeIdx, From, Type, Options, Owners) ->
@ -596,7 +596,7 @@
%% @spec (Host, Node, NodeId, Type, LJID, Number) -> any() %% @spec (Host, Node, NodeId, Type, LJID, Number) -> any()
@@ -2318,16 +2141,27 @@ @@ -2313,16 +2136,27 @@
%% @doc <p>Resend the items of a node to the user.</p> %% @doc <p>Resend the items of a node to the user.</p>
%% @todo use cache-last-item feature %% @todo use cache-last-item feature
send_items(Host, Node, NodeId, Type, LJID, last) -> send_items(Host, Node, NodeId, Type, LJID, last) ->
@ -630,7 +630,7 @@
send_items(Host, Node, NodeId, Type, LJID, Number) -> send_items(Host, Node, NodeId, Type, LJID, Number) ->
ToSend = case node_action(Host, Type, get_items, [NodeId, LJID]) of ToSend = case node_action(Host, Type, get_items, [NodeId, LJID]) of
{result, []} -> {result, []} ->
@@ -2453,7 +2287,8 @@ @@ -2448,7 +2282,8 @@
error -> error ->
{error, ?ERR_BAD_REQUEST}; {error, ?ERR_BAD_REQUEST};
_ -> _ ->
@ -640,7 +640,7 @@
case lists:member(Owner, Owners) of case lists:member(Owner, Owners) of
true -> true ->
OwnerJID = jlib:make_jid(Owner), OwnerJID = jlib:make_jid(Owner),
@@ -2463,24 +2298,7 @@ @@ -2458,24 +2293,7 @@
end, end,
lists:foreach( lists:foreach(
fun({JID, Affiliation}) -> fun({JID, Affiliation}) ->
@ -666,7 +666,7 @@
end, FilteredEntities), end, FilteredEntities),
{result, []}; {result, []};
_ -> _ ->
@@ -2533,11 +2351,11 @@ @@ -2528,11 +2346,11 @@
end. end.
read_sub(Subscriber, Node, NodeID, SubID, Lang) -> read_sub(Subscriber, Node, NodeID, SubID, Lang) ->
@ -680,7 +680,7 @@
OptionsEl = {xmlelement, "options", [{"jid", jlib:jid_to_string(Subscriber)}, OptionsEl = {xmlelement, "options", [{"jid", jlib:jid_to_string(Subscriber)},
{"subid", SubID}|nodeAttr(Node)], {"subid", SubID}|nodeAttr(Node)],
[XdataEl]}, [XdataEl]},
@@ -2563,7 +2381,7 @@ @@ -2558,7 +2376,7 @@
end. end.
set_options_helper(Configuration, JID, NodeID, SubID, Type) -> set_options_helper(Configuration, JID, NodeID, SubID, Type) ->
@ -689,7 +689,7 @@
{result, GoodSubOpts} -> GoodSubOpts; {result, GoodSubOpts} -> GoodSubOpts;
_ -> invalid _ -> invalid
end, end,
@@ -2592,7 +2410,7 @@ @@ -2587,7 +2405,7 @@
write_sub(_Subscriber, _NodeID, _SubID, invalid) -> write_sub(_Subscriber, _NodeID, _SubID, invalid) ->
{error, extended_error(?ERR_BAD_REQUEST, "invalid-options")}; {error, extended_error(?ERR_BAD_REQUEST, "invalid-options")};
write_sub(Subscriber, NodeID, SubID, Options) -> write_sub(Subscriber, NodeID, SubID, Options) ->
@ -698,7 +698,7 @@
{error, notfound} -> {error, notfound} ->
{error, extended_error(?ERR_NOT_ACCEPTABLE, "invalid-subid")}; {error, extended_error(?ERR_NOT_ACCEPTABLE, "invalid-subid")};
{result, _} -> {result, _} ->
@@ -2760,8 +2578,8 @@ @@ -2755,8 +2573,8 @@
{"subscription", subscription_to_string(Sub)} | nodeAttr(Node)], []}]}]}, {"subscription", subscription_to_string(Sub)} | nodeAttr(Node)], []}]}]},
ejabberd_router:route(service_jid(Host), jlib:make_jid(JID), Stanza) ejabberd_router:route(service_jid(Host), jlib:make_jid(JID), Stanza)
end, end,
@ -709,28 +709,27 @@
true -> true ->
Result = lists:foldl(fun({JID, Subscription, SubId}, Acc) -> Result = lists:foldl(fun({JID, Subscription, SubId}, Acc) ->
@@ -3116,7 +2934,7 @@ @@ -3119,7 +2937,7 @@
{Depth, [{N, get_node_subs(N)} || N <- Nodes]} Collection = tree_call(Host, get_parentnodes_tree, [Host, Node, service_jid(Host)]),
end, tree_call(Host, get_parentnodes_tree, [Host, Node, service_jid(Host)]))} {result, [{Depth, [{N, sub_with_options(N)} || N <- Nodes]} || {Depth, Nodes} <- Collection]}
end, end,
- case transaction(Action, sync_dirty) of - case transaction(Action, sync_dirty) of
+ case transaction(Host, Action, sync_dirty) of + case transaction(Host, Action, sync_dirty) of
{result, CollSubs} -> CollSubs; {result, CollSubs} -> subscribed_nodes_by_jid(NotifyType, CollSubs);
_ -> [] _ -> []
end. end.
@@ -3130,9 +2948,9 @@ @@ -3188,8 +3006,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:get_subscription(JID, NodeId, SubId) of
+ {result, #pubsub_subscription{options = Options}} -> {JID, SubId, Options};
_ -> {JID, SubId, []}
end.
get_options_for_subs(NodeID, Subs) -> @@ -3265,6 +3083,30 @@
lists:foldl(fun({JID, subscribed, SubID}, Acc) ->
- case pubsub_subscription:read_subscription(JID, NodeID, SubID) of
+ case pubsub_subscription_odbc:get_subscription(JID, NodeID, SubID) of
{error, notfound} -> [{JID, SubID, []} | Acc];
- #pubsub_subscription{options = Options} -> [{JID, SubID, Options} | Acc];
+ {result, #pubsub_subscription{options = Options}} -> [{JID, SubID, Options} | Acc];
_ -> Acc
end;
(_, Acc) ->
@@ -3321,6 +3139,30 @@
Result Result
end. end.
@ -761,7 +760,7 @@
%% @spec (Host, Options) -> MaxItems %% @spec (Host, Options) -> MaxItems
%% Host = host() %% Host = host()
%% Options = [Option] %% Options = [Option]
@@ -3717,7 +3559,13 @@ @@ -3661,7 +3503,13 @@
tree_action(Host, Function, Args) -> tree_action(Host, Function, Args) ->
?DEBUG("tree_action ~p ~p ~p",[Host,Function,Args]), ?DEBUG("tree_action ~p ~p ~p",[Host,Function,Args]),
Fun = fun() -> tree_call(Host, Function, Args) end, Fun = fun() -> tree_call(Host, Function, Args) end,
@ -776,7 +775,7 @@
%% @doc <p>node plugin call.</p> %% @doc <p>node plugin call.</p>
node_call(Type, Function, Args) -> node_call(Type, Function, Args) ->
@@ -3737,13 +3585,13 @@ @@ -3681,13 +3529,13 @@
node_action(Host, Type, Function, Args) -> node_action(Host, Type, Function, Args) ->
?DEBUG("node_action ~p ~p ~p ~p",[Host,Type,Function,Args]), ?DEBUG("node_action ~p ~p ~p ~p",[Host,Type,Function,Args]),
@ -792,7 +791,7 @@
case tree_call(Host, get_node, [Host, Node]) of case tree_call(Host, get_node, [Host, Node]) of
N when is_record(N, pubsub_node) -> N when is_record(N, pubsub_node) ->
case Action(N) of case Action(N) of
@@ -3755,13 +3603,19 @@ @@ -3699,13 +3547,19 @@
Error Error
end end
end, Trans). end, Trans).
@ -816,7 +815,7 @@
{result, Result} -> {result, Result}; {result, Result} -> {result, Result};
{error, Error} -> {error, Error}; {error, Error} -> {error, Error};
{atomic, {result, Result}} -> {result, Result}; {atomic, {result, Result}} -> {result, Result};
@@ -3769,6 +3623,15 @@ @@ -3713,6 +3567,15 @@
{aborted, Reason} -> {aborted, Reason} ->
?ERROR_MSG("transaction return internal error: ~p~n", [{aborted, Reason}]), ?ERROR_MSG("transaction return internal error: ~p~n", [{aborted, Reason}]),
{error, ?ERR_INTERNAL_SERVER_ERROR}; {error, ?ERR_INTERNAL_SERVER_ERROR};
@ -832,7 +831,7 @@
{'EXIT', Reason} -> {'EXIT', Reason} ->
?ERROR_MSG("transaction return internal error: ~p~n", [{'EXIT', Reason}]), ?ERROR_MSG("transaction return internal error: ~p~n", [{'EXIT', Reason}]),
{error, ?ERR_INTERNAL_SERVER_ERROR}; {error, ?ERR_INTERNAL_SERVER_ERROR};
@@ -3777,6 +3640,17 @@ @@ -3721,6 +3584,17 @@
{error, ?ERR_INTERNAL_SERVER_ERROR} {error, ?ERR_INTERNAL_SERVER_ERROR}
end. end.