--- mod_pubsub.erl 2010-06-02 15:03:48.000000000 +0200 +++ mod_pubsub_odbc.erl 2010-06-02 16:45:38.000000000 +0200 @@ -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. --module(mod_pubsub). +-module(mod_pubsub_odbc). -author('christophe.romain@process-one.net'). -version('1.13-0'). @@ -54,9 +54,9 @@ -include("jlib.hrl"). -include("pubsub.hrl"). --define(STDTREE, "tree"). --define(STDNODE, "flat"). --define(PEPNODE, "pep"). +-define(STDTREE, "tree_odbc"). +-define(STDNODE, "flat_odbc"). +-define(PEPNODE, "pep_odbc"). %% exports for hooks -export([presence_probe/3, @@ -104,7 +104,7 @@ string_to_affiliation/1, extended_error/2, extended_error/3, - rename_default_nodeplugin/0 + escape/1 ]). %% API and gen_server callbacks @@ -123,7 +123,7 @@ -export([send_loop/1 ]). --define(PROCNAME, ejabberd_mod_pubsub). +-define(PROCNAME, ejabberd_mod_pubsub_odbc). -define(LOOPNAME, ejabberd_mod_pubsub_loop). -define(PLUGIN_PREFIX, "node_"). -define(TREE_PREFIX, "nodetree_"). @@ -220,8 +220,6 @@ ok end, ejabberd_router:register_route(Host), - update_node_database(Host, ServerHost), - update_state_database(Host, ServerHost), init_nodes(Host, ServerHost, NodeTree, Plugins), State = #state{host = Host, server_host = ServerHost, @@ -280,207 +278,14 @@ init_nodes(Host, ServerHost, _NodeTree, Plugins) -> %% TODO, this call should be done plugin side - case lists:member("hometree", Plugins) of + case lists:member("hometree_odbc", Plugins) of true -> - create_node(Host, ServerHost, string_to_node("/home"), service_jid(Host), "hometree"), - create_node(Host, ServerHost, string_to_node("/home/"++ServerHost), service_jid(Host), "hometree"); + create_node(Host, ServerHost, string_to_node("/home"), service_jid(Host), "hometree_odbc"), + create_node(Host, ServerHost, string_to_node("/home/"++ServerHost), service_jid(Host), "hometree_odbc"); false -> ok end. -update_node_database(Host, ServerHost) -> - mnesia:del_table_index(pubsub_node, type), - mnesia:del_table_index(pubsub_node, parentid), - case catch mnesia:table_info(pubsub_node, attributes) of - [host_node, host_parent, info] -> - ?INFO_MSG("upgrade node pubsub tables",[]), - F = fun() -> - {Result, LastIdx} = lists:foldl( - fun({pubsub_node, NodeId, ParentId, {nodeinfo, Items, Options, Entities}}, {RecList, NodeIdx}) -> - ItemsList = - lists:foldl( - fun({item, IID, Publisher, Payload}, Acc) -> - C = {unknown, Publisher}, - M = {now(), Publisher}, - mnesia:write( - #pubsub_item{itemid = {IID, NodeIdx}, - creation = C, - modification = M, - payload = Payload}), - [{Publisher, IID} | Acc] - end, [], Items), - Owners = - dict:fold( - fun(JID, {entity, Aff, Sub}, Acc) -> - UsrItems = - lists:foldl( - fun({P, I}, IAcc) -> - case P of - JID -> [I | IAcc]; - _ -> IAcc - end - end, [], ItemsList), - mnesia:write({pubsub_state, - {JID, NodeIdx}, - UsrItems, - Aff, - Sub}), - case Aff of - owner -> [JID | Acc]; - _ -> Acc - end - end, [], Entities), - mnesia:delete({pubsub_node, NodeId}), - {[#pubsub_node{nodeid = NodeId, - id = NodeIdx, - parents = [element(2, ParentId)], - owners = Owners, - options = Options} | - RecList], NodeIdx + 1} - end, {[], 1}, - mnesia:match_object( - {pubsub_node, {Host, '_'}, '_', '_'})), - mnesia:write(#pubsub_index{index = node, last = LastIdx, free = []}), - Result - end, - {atomic, NewRecords} = mnesia:transaction(F), - {atomic, ok} = mnesia:delete_table(pubsub_node), - {atomic, ok} = mnesia:create_table(pubsub_node, - [{disc_copies, [node()]}, - {attributes, record_info(fields, pubsub_node)}]), - FNew = fun() -> lists:foreach(fun(Record) -> - mnesia:write(Record) - end, NewRecords) - end, - case mnesia:transaction(FNew) of - {atomic, Result} -> - ?INFO_MSG("Pubsub node tables updated correctly: ~p", [Result]); - {aborted, Reason} -> - ?ERROR_MSG("Problem updating Pubsub node tables:~n~p", [Reason]) - end; - [nodeid, parentid, type, owners, options] -> - F = fun({pubsub_node, NodeId, {_, Parent}, Type, Owners, Options}) -> - #pubsub_node{ - nodeid = NodeId, - id = 0, - parents = [Parent], - type = Type, - owners = Owners, - options = Options} - end, - mnesia:transform_table(pubsub_node, F, [nodeid, id, parents, type, owners, options]), - FNew = fun() -> - LastIdx = lists:foldl(fun(#pubsub_node{nodeid = NodeId} = PubsubNode, NodeIdx) -> - mnesia:write(PubsubNode#pubsub_node{id = NodeIdx}), - lists:foreach(fun(#pubsub_state{stateid = StateId} = State) -> - {JID, _} = StateId, - mnesia:delete({pubsub_state, StateId}), - mnesia:write(State#pubsub_state{stateid = {JID, NodeIdx}}) - end, mnesia:match_object(#pubsub_state{stateid = {'_', NodeId}, _ = '_'})), - lists:foreach(fun(#pubsub_item{itemid = ItemId} = Item) -> - {IID, _} = ItemId, - {M1, M2} = Item#pubsub_item.modification, - {C1, C2} = Item#pubsub_item.creation, - mnesia:delete({pubsub_item, ItemId}), - mnesia:write(Item#pubsub_item{itemid = {IID, NodeIdx}, - modification = {M2, M1}, - creation = {C2, C1}}) - end, mnesia:match_object(#pubsub_item{itemid = {'_', NodeId}, _ = '_'})), - NodeIdx + 1 - end, 1, mnesia:match_object( - {pubsub_node, {Host, '_'}, '_', '_', '_', '_', '_'}) - ++ mnesia:match_object( - {pubsub_node, {{'_', ServerHost, '_'}, '_'}, '_', '_', '_', '_', '_'})), - mnesia:write(#pubsub_index{index = node, last = LastIdx, free = []}) - end, - case mnesia:transaction(FNew) of - {atomic, Result} -> - rename_default_nodeplugin(), - ?INFO_MSG("Pubsub node tables updated correctly: ~p", [Result]); - {aborted, Reason} -> - ?ERROR_MSG("Problem updating Pubsub node tables:~n~p", [Reason]) - end; - [nodeid, id, parent, type, owners, options] -> - F = fun({pubsub_node, NodeId, Id, Parent, Type, Owners, Options}) -> - #pubsub_node{ - nodeid = NodeId, - id = Id, - parents = [Parent], - type = Type, - owners = Owners, - options = Options} - end, - mnesia:transform_table(pubsub_node, F, [nodeid, id, parents, type, owners, options]), - rename_default_nodeplugin(); - _ -> - ok - end, - mnesia:transaction(fun() -> - case catch mnesia:first(pubsub_node) of - {_, L} when is_list(L) -> - lists:foreach( - fun({H, N}) when is_list(N) -> - [Node] = mnesia:read({pubsub_node, {H, N}}), - Type = Node#pubsub_node.type, - BN = element(2, node_call(Type, path_to_node, [N])), - BP = case [element(2, node_call(Type, path_to_node, [P])) || P <- Node#pubsub_node.parents] of - [<<>>] -> []; - Parents -> Parents - end, - mnesia:write(Node#pubsub_node{nodeid={H, BN}, parents=BP}), - mnesia:delete({pubsub_node, {H, N}}); - (_) -> - ok - end, mnesia:all_keys(pubsub_node)); - _ -> - ok - end - end). - -rename_default_nodeplugin() -> - lists:foreach(fun(Node) -> - mnesia:dirty_write(Node#pubsub_node{type = "hometree"}) - end, mnesia:dirty_match_object(#pubsub_node{type = "default", _ = '_'})). - -update_state_database(_Host, _ServerHost) -> - case catch mnesia:table_info(pubsub_state, attributes) of - [stateid, items, affiliation, subscription] -> - ?INFO_MSG("upgrade state pubsub tables", []), - F = fun ({pubsub_state, {JID, NodeID}, Items, Aff, Sub}, Acc) -> - Subs = case Sub of - none -> - []; - _ -> - {result, SubID} = pubsub_subscription:subscribe_node(JID, NodeID, []), - [{Sub, SubID}] - end, - NewState = #pubsub_state{stateid = {JID, NodeID}, - items = Items, - affiliation = Aff, - subscriptions = Subs}, - [NewState | Acc] - end, - {atomic, NewRecs} = mnesia:transaction(fun mnesia:foldl/3, - [F, [], pubsub_state]), - {atomic, ok} = mnesia:delete_table(pubsub_state), - {atomic, ok} = mnesia:create_table(pubsub_state, - [{disc_copies, [node()]}, - {attributes, record_info(fields, pubsub_state)}]), - FNew = fun () -> - lists:foreach(fun mnesia:write/1, NewRecs) - end, - case mnesia:transaction(FNew) of - {atomic, Result} -> - ?INFO_MSG("Pubsub state tables updated correctly: ~p", - [Result]); - {aborted, Reason} -> - ?ERROR_MSG("Problem updating Pubsub state tables:~n~p", - [Reason]) - end; - _ -> - ok - end. - send_loop(State) -> receive {presence, JID, Pid} -> @@ -491,17 +296,15 @@ %% 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) -> - {result, Subscriptions} = node_action(Host, PType, get_entity_subscriptions, [Host, JID]), + Subscriptions = case catch node_action(Host, PType, get_entity_subscriptions_for_send_last, [Host, JID]) of + {result, S} -> S; + _ -> [] + end, lists:foreach( fun({Node, subscribed, _, SubJID}) -> if (SubJID == LJID) or (SubJID == BJID) -> - #pubsub_node{nodeid = {H, N}, type = Type, id = NodeId, options = Options} = Node, - case get_option(Options, send_last_published_item) of - on_sub_and_presence -> - send_items(H, N, NodeId, Type, LJID, last); - _ -> - ok - end; + #pubsub_node{nodeid = {H, N}, type = Type, id = NodeId} = Node, + send_items(H, N, NodeId, Type, LJID, last); true -> % resource not concerned about that subscription ok @@ -747,10 +550,10 @@ lists:foreach(fun(PType) -> {result, Subscriptions} = node_action(Host, PType, get_entity_subscriptions, [Host, Entity]), lists:foreach(fun - ({#pubsub_node{options = Options, owners = Owners, id = NodeId}, subscribed, _, JID}) -> + ({#pubsub_node{options = Options, id = NodeId}, subscribed, _, JID}) -> case get_option(Options, access_model) of presence -> - case lists:member(BJID, Owners) of + case lists:member(BJID, node_owners(Host, PType, NodeId)) of true -> node_action(Host, PType, unsubscribe_node, [NodeId, Entity, JID, all]); false -> @@ -920,7 +723,8 @@ sub_el = SubEl} = IQ -> {xmlelement, _, QAttrs, _} = SubEl, Node = xml:get_attr_s("node", QAttrs), - Res = case iq_disco_items(Host, Node, From) of + Rsm = jlib:rsm_decode(IQ), + Res = case iq_disco_items(Host, Node, From, Rsm) of {result, IQRes} -> jlib:iq_to_xml( IQ#iq{type = result, @@ -1033,7 +837,7 @@ [] -> ["leaf"]; %% No sub-nodes: it's a leaf node _ -> - case node_call(Type, get_items, [NodeId, From]) of + case node_call(Type, get_items, [NodeId, From, none]) of {result, []} -> ["collection"]; {result, _} -> ["leaf", "collection"]; _ -> [] @@ -1049,8 +853,9 @@ []; true -> [{xmlelement, "feature", [{"var", ?NS_PUBSUB}], []} | - lists:map(fun(T) -> - {xmlelement, "feature", [{"var", ?NS_PUBSUB++"#"++T}], []} + lists:map(fun + ("rsm")-> {xmlelement, "feature", [{"var", ?NS_RSM}], []}; + (T) -> {xmlelement, "feature", [{"var", ?NS_PUBSUB++"#"++T}], []} end, features(Type))] end, %% TODO: add meta-data info (spec section 5.4) @@ -1079,8 +884,9 @@ {xmlelement, "feature", [{"var", ?NS_PUBSUB}], []}, {xmlelement, "feature", [{"var", ?NS_COMMANDS}], []}, {xmlelement, "feature", [{"var", ?NS_VCARD}], []}] ++ - lists:map(fun(Feature) -> - {xmlelement, "feature", [{"var", ?NS_PUBSUB++"#"++Feature}], []} + lists:map(fun + ("rsm")-> {xmlelement, "feature", [{"var", ?NS_RSM}], []}; + (T) -> {xmlelement, "feature", [{"var", ?NS_PUBSUB++"#"++T}], []} end, features(Host, Node))}; <> -> command_disco_info(Host, Node, From); @@ -1090,7 +896,7 @@ node_disco_info(Host, Node, From) end. -iq_disco_items(Host, [], From) -> +iq_disco_items(Host, [], From, _RSM) -> case tree_action(Host, get_subnodes, [Host, <<>>, From]) of Nodes when is_list(Nodes) -> {result, lists:map( @@ -1107,14 +913,14 @@ Other -> Other end; -iq_disco_items(Host, ?NS_COMMANDS, _From) -> +iq_disco_items(Host, ?NS_COMMANDS, _From, _RSM) -> %% TODO: support localization of this string CommandItems = [{xmlelement, "item", [{"jid", Host}, {"node", ?NS_PUBSUB_GET_PENDING}, {"name", "Get Pending"}], []}], {result, CommandItems}; -iq_disco_items(_Host, ?NS_PUBSUB_GET_PENDING, _From) -> +iq_disco_items(_Host, ?NS_PUBSUB_GET_PENDING, _From, _RSM) -> CommandItems = [], {result, CommandItems}; -iq_disco_items(Host, Item, From) -> +iq_disco_items(Host, Item, From, RSM) -> case string:tokens(Item, "!") of [_SNode, _ItemID] -> {result, []}; @@ -1122,10 +928,10 @@ Node = string_to_node(SNode), Action = fun(#pubsub_node{type = Type, id = NodeId}) -> - % TODO call get_items/6 instead for access control (EJAB-1033) - NodeItems = case node_call(Type, get_items, [NodeId, From]) of + %% TODO call get_items/6 instead for access control (EJAB-1033) + {NodeItems, RsmOut} = case node_call(Type, get_items, [NodeId, From, RSM]) of {result, I} -> I; - _ -> [] + _ -> {[], none} end, Nodes = lists:map( fun(#pubsub_node{nodeid = {_, SubNode}, options = Options}) -> @@ -1143,7 +949,7 @@ {result, Name} = node_call(Type, get_item_name, [Host, Node, RN]), {xmlelement, "item", [{"jid", Host}, {"name", Name}], []} end, NodeItems), - {result, Nodes ++ Items} + {result, Nodes ++ Items ++ jlib:rsm_encode(RsmOut)} end, case transaction(Host, Node, Action, sync_dirty) of {result, {_, Result}} -> {result, Result}; @@ -1272,7 +1078,8 @@ (_, Acc) -> Acc end, [], xml:remove_cdata(Els)), - get_items(Host, Node, From, SubId, MaxItems, ItemIDs); + RSM = jlib:rsm_decode(SubEl), + get_items(Host, Node, From, SubId, MaxItems, ItemIDs, RSM); {get, "subscriptions"} -> get_subscriptions(Host, Node, From, Plugins); {get, "affiliations"} -> @@ -1295,7 +1102,9 @@ iq_pubsub_owner(Host, ServerHost, From, IQType, SubEl, Lang) -> {xmlelement, _, _, SubEls} = SubEl, - Action = xml:remove_cdata(SubEls), + Action = lists:filter(fun({xmlelement, "set", _, _}) -> false; + (_) -> true + end, xml:remove_cdata(SubEls)), case Action of [{xmlelement, Name, Attrs, Els}] -> Node = string_to_node(xml:get_attr_s("node", Attrs)), @@ -1425,7 +1234,8 @@ _ -> [] end end, - case transaction(fun () -> {result, lists:flatmap(Tr, Plugins)} end, + case transaction(Host, + fun () -> {result, lists:flatmap(Tr, Plugins)} end, sync_dirty) of {result, Res} -> Res; Err -> Err @@ -1464,7 +1274,7 @@ %%% authorization handling -send_authorization_request(#pubsub_node{owners = Owners, nodeid = {Host, Node}}, Subscriber) -> +send_authorization_request(#pubsub_node{nodeid = {Host, Node}, type = Type, id = NodeId}, Subscriber) -> Lang = "en", %% TODO fix Stanza = {xmlelement, "message", [], @@ -1493,7 +1303,7 @@ [{xmlelement, "value", [], [{xmlcdata, "false"}]}]}]}]}, lists:foreach(fun(Owner) -> ejabberd_router:route(service_jid(Host), jlib:make_jid(Owner), Stanza) - end, Owners). + end, node_owners(Host, Type, NodeId)). find_authorization_response(Packet) -> {xmlelement, _Name, _Attrs, Els} = Packet, @@ -1557,8 +1367,8 @@ "true" -> true; _ -> false end, - Action = fun(#pubsub_node{type = Type, owners = Owners, id = NodeId}) -> - IsApprover = lists:member(jlib:jid_tolower(jlib:jid_remove_resource(From)), Owners), + Action = fun(#pubsub_node{type = Type, id = NodeId}) -> + IsApprover = lists:member(jlib:jid_tolower(jlib:jid_remove_resource(From)), node_owners_call(Type, NodeId)), {result, Subscriptions} = node_call(Type, get_subscriptions, [NodeId, Subscriber]), if not IsApprover -> @@ -1757,7 +1567,7 @@ Reply = [{xmlelement, "pubsub", [{"xmlns", ?NS_PUBSUB}], [{xmlelement, "create", 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), case Result of @@ -1860,7 +1670,7 @@ %%
  • The node does not exist.
  • %% subscribe_node(Host, Node, From, JID, Configuration) -> - SubOpts = case pubsub_subscription:parse_options_xform(Configuration) of + SubOpts = case pubsub_subscription_odbc:parse_options_xform(Configuration) of {result, GoodSubOpts} -> GoodSubOpts; _ -> invalid end, @@ -1868,7 +1678,7 @@ error -> {"", "", ""}; J -> jlib:jid_tolower(J) end, - Action = fun(#pubsub_node{options = Options, owners = [Owner|_], type = Type, id = NodeId}) -> + Action = fun(#pubsub_node{options = Options, type = Type, id = NodeId}) -> Features = features(Type), SubscribeFeature = lists:member("subscribe", Features), OptionsFeature = lists:member("subscription-options", Features), @@ -1887,9 +1697,13 @@ {"", "", ""} -> {false, false}; _ -> - {OU, OS, _} = Owner, - get_roster_info(OU, OS, - Subscriber, AllowedGroups) + case node_owners_call(Type, NodeId) of + [{OU, OS, _}|_] -> + get_roster_info(OU, OS, + Subscriber, AllowedGroups); + _ -> + {false, false} + end end end, if @@ -2220,7 +2034,7 @@ %%

    The permission are not checked in this function.

    %% @todo We probably need to check that the user doing the query has the right %% to read the items. -get_items(Host, Node, From, SubId, SMaxItems, ItemIDs) -> +get_items(Host, Node, From, SubId, SMaxItems, ItemIDs, RSM) -> MaxItems = if SMaxItems == "" -> get_max_items_node(Host); @@ -2259,11 +2073,11 @@ node_call(Type, get_items, [NodeId, From, AccessModel, PresenceSubscription, RosterGroup, - SubId]) + SubId, RSM]) end end, case transaction(Host, Node, Action, sync_dirty) of - {result, {_, Items}} -> + {result, {_, {Items, RSMOut}}} -> SendItems = case ItemIDs of [] -> Items; @@ -2276,7 +2090,8 @@ %% number of items sent to MaxItems: {result, [{xmlelement, "pubsub", [{"xmlns", ?NS_PUBSUB}], [{xmlelement, "items", nodeAttr(Node), - itemsEls(lists:sublist(SendItems, MaxItems))}]}]}; + itemsEls(lists:sublist(SendItems, MaxItems))} + | jlib:rsm_encode(RSMOut)]}]}; Error -> Error end @@ -2308,16 +2123,27 @@ %% @doc

    Resend the items of a node to the user.

    %% @todo use cache-last-item feature send_items(Host, Node, NodeId, Type, LJID, last) -> - case get_cached_item(Host, NodeId) of + Stanza = case get_cached_item(Host, NodeId) of undefined -> - send_items(Host, Node, NodeId, Type, LJID, 1); + % special ODBC optimization, works only with node_hometree_odbc, node_flat_odbc and node_pep_odbc + case node_action(Host, Type, get_last_items, [NodeId, LJID, 1]) of + {result, [LastItem]} -> + {ModifNow, ModifUSR} = LastItem#pubsub_item.modification, + event_stanza_with_delay( + [{xmlelement, "items", nodeAttr(Node), + itemsEls([LastItem])}], ModifNow, ModifUSR); + _ -> + event_stanza( + [{xmlelement, "items", nodeAttr(Node), + itemsEls([])}]) + end; LastItem -> {ModifNow, ModifUSR} = LastItem#pubsub_item.modification, - Stanza = event_stanza_with_delay( + event_stanza_with_delay( [{xmlelement, "items", nodeAttr(Node), - itemsEls([LastItem])}], ModifNow, ModifUSR), - ejabberd_router:route(service_jid(Host), jlib:make_jid(LJID), Stanza) - end; + itemsEls([LastItem])}], ModifNow, ModifUSR) + end, + ejabberd_router:route(service_jid(Host), jlib:make_jid(LJID), Stanza); send_items(Host, Node, NodeId, Type, LJID, Number) -> ToSend = case node_action(Host, Type, get_items, [NodeId, LJID]) of {result, []} -> @@ -2443,7 +2269,8 @@ error -> {error, ?ERR_BAD_REQUEST}; _ -> - Action = fun(#pubsub_node{owners = Owners, type = Type, id = NodeId}=N) -> + Action = fun(#pubsub_node{type = Type, id = NodeId}) -> + Owners = node_owners_call(Type, NodeId), case lists:member(Owner, Owners) of true -> OwnerJID = jlib:make_jid(Owner), @@ -2453,24 +2280,7 @@ end, lists:foreach( fun({JID, Affiliation}) -> - node_call(Type, set_affiliation, [NodeId, JID, Affiliation]), - case Affiliation of - owner -> - NewOwner = jlib:jid_tolower(jlib:jid_remove_resource(JID)), - NewOwners = [NewOwner|Owners], - tree_call(Host, set_node, [N#pubsub_node{owners = NewOwners}]); - none -> - OldOwner = jlib:jid_tolower(jlib:jid_remove_resource(JID)), - case lists:member(OldOwner, Owners) of - true -> - NewOwners = Owners--[OldOwner], - tree_call(Host, set_node, [N#pubsub_node{owners = NewOwners}]); - _ -> - ok - end; - _ -> - ok - end + node_call(Type, set_affiliation, [NodeId, JID, Affiliation]) end, FilteredEntities), {result, []}; _ -> @@ -2523,11 +2333,11 @@ end. read_sub(Subscriber, Node, NodeID, SubID, Lang) -> - case pubsub_subscription:get_subscription(Subscriber, NodeID, SubID) of + case pubsub_subscription_odbc:get_subscription(Subscriber, NodeID, SubID) of {error, notfound} -> {error, extended_error(?ERR_NOT_ACCEPTABLE, "invalid-subid")}; {result, #pubsub_subscription{options = Options}} -> - {result, XdataEl} = pubsub_subscription:get_options_xform(Lang, Options), + {result, XdataEl} = pubsub_subscription_odbc:get_options_xform(Lang, Options), OptionsEl = {xmlelement, "options", [{"jid", jlib:jid_to_string(Subscriber)}, {"subid", SubID}|nodeAttr(Node)], [XdataEl]}, @@ -2553,7 +2363,7 @@ end. set_options_helper(Configuration, JID, NodeID, SubID, Type) -> - SubOpts = case pubsub_subscription:parse_options_xform(Configuration) of + SubOpts = case pubsub_subscription_odbc:parse_options_xform(Configuration) of {result, GoodSubOpts} -> GoodSubOpts; _ -> invalid end, @@ -2582,7 +2392,7 @@ write_sub(_Subscriber, _NodeID, _SubID, invalid) -> {error, extended_error(?ERR_BAD_REQUEST, "invalid-options")}; write_sub(Subscriber, NodeID, SubID, Options) -> - case pubsub_subscription:set_subscription(Subscriber, NodeID, SubID, Options) of + case pubsub_subscription_odbc:set_subscription(Subscriber, NodeID, SubID, Options) of {error, notfound} -> {error, extended_error(?ERR_NOT_ACCEPTABLE, "invalid-subid")}; {result, _} -> @@ -2750,8 +2560,8 @@ {"subscription", subscription_to_string(Sub)} | nodeAttr(Node)], []}]}]}, ejabberd_router:route(service_jid(Host), jlib:make_jid(JID), Stanza) end, - Action = fun(#pubsub_node{owners = Owners, type = Type, id = NodeId}) -> - case lists:member(Owner, Owners) of + Action = fun(#pubsub_node{type = Type, id = NodeId}) -> + case lists:member(Owner, node_owners_call(Type, NodeId)) of true -> Result = lists:foldl(fun({JID, Subscription, SubId}, Acc) -> @@ -3088,7 +2898,7 @@ {Depth, [{N, get_node_subs(N)} || N <- Nodes]} end, tree_call(Host, get_parentnodes_tree, [Host, Node, service_jid(Host)]))} end, - case transaction(Action, sync_dirty) of + case transaction(Host, Action, sync_dirty) of {result, CollSubs} -> CollSubs; _ -> [] end. @@ -3102,9 +2912,9 @@ get_options_for_subs(NodeID, Subs) -> 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) -> @@ -3308,6 +3118,30 @@ Result end. +%% @spec (Host, Type, NodeId) -> [ljid()] +%% NodeId = pubsubNodeId() +%% @doc

    Return list of node owners.

    +node_owners(Host, Type, NodeId) -> + case node_action(Host, Type, get_node_affiliations, [NodeId]) of + {result, Affiliations} -> + lists:foldl( + fun({LJID, owner}, Acc) -> [LJID|Acc]; + (_, Acc) -> Acc + end, [], Affiliations); + _ -> + [] + end. +node_owners_call(Type, NodeId) -> + case node_call(Type, get_node_affiliations, [NodeId]) of + {result, Affiliations} -> + lists:foldl( + fun({LJID, owner}, Acc) -> [LJID|Acc]; + (_, Acc) -> Acc + end, [], Affiliations); + _ -> + [] + end. + %% @spec (Host, Options) -> MaxItems %% Host = host() %% Options = [Option] @@ -3704,7 +3538,13 @@ tree_action(Host, Function, Args) -> ?DEBUG("tree_action ~p ~p ~p",[Host,Function,Args]), Fun = fun() -> tree_call(Host, Function, Args) end, - catch mnesia:sync_dirty(Fun). + case catch ejabberd_odbc:sql_bloc(odbc_conn(Host), Fun) of + {atomic, Result} -> + Result; + {aborted, Reason} -> + ?ERROR_MSG("transaction return internal error: ~p~n",[{aborted, Reason}]), + {error, ?ERR_INTERNAL_SERVER_ERROR} + end. %% @doc

    node plugin call.

    node_call(Type, Function, Args) -> @@ -3724,13 +3564,13 @@ node_action(Host, Type, Function, Args) -> ?DEBUG("node_action ~p ~p ~p ~p",[Host,Type,Function,Args]), - transaction(fun() -> + transaction(Host, fun() -> node_call(Type, Function, Args) end, sync_dirty). %% @doc

    plugin transaction handling.

    transaction(Host, Node, Action, Trans) -> - transaction(fun() -> + transaction(Host, fun() -> case tree_call(Host, get_node, [Host, Node]) of N when is_record(N, pubsub_node) -> case Action(N) of @@ -3743,8 +3583,14 @@ end end, Trans). -transaction(Fun, Trans) -> - case catch mnesia:Trans(Fun) of +transaction(Host, Fun, Trans) -> + transaction_retry(Host, Fun, Trans, 2). +transaction_retry(Host, Fun, Trans, Count) -> + SqlFun = case Trans of + transaction -> sql_transaction; + _ -> sql_bloc + end, + case catch ejabberd_odbc:SqlFun(odbc_conn(Host), Fun) of {result, Result} -> {result, Result}; {error, Error} -> {error, Error}; {atomic, {result, Result}} -> {result, Result}; @@ -3752,6 +3598,15 @@ {aborted, Reason} -> ?ERROR_MSG("transaction return internal error: ~p~n", [{aborted, Reason}]), {error, ?ERR_INTERNAL_SERVER_ERROR}; + {'EXIT', {timeout, _} = Reason} -> + case Count of + 0 -> + ?ERROR_MSG("transaction return internal error: ~p~n", [{'EXIT', Reason}]), + {error, ?ERR_INTERNAL_SERVER_ERROR}; + N -> + erlang:yield(), + transaction_retry(Host, Fun, Trans, N-1) + end; {'EXIT', Reason} -> ?ERROR_MSG("transaction return internal error: ~p~n", [{'EXIT', Reason}]), {error, ?ERR_INTERNAL_SERVER_ERROR}; @@ -3760,6 +3615,17 @@ {error, ?ERR_INTERNAL_SERVER_ERROR} end. +odbc_conn({_U, Host, _R})-> + Host; +odbc_conn(Host) -> + lists:dropwhile(fun(A) -> A/=$. end, Host) -- ".". + +%% escape value for database storage +escape({_U, _H, _R}=JID)-> + ejabberd_odbc:escape(jlib:jid_to_string(JID)); +escape(Value)-> + ejabberd_odbc:escape(Value). + %%%% helpers %% Add pubsub-specific error element