Additional pub-sub work.

git-svn-id: http://svn.igniterealtime.org/svn/repos/wildfire/trunk@3639 b35dd754-fafc-0310-a699-88a17e54d16e

src/java/org/jivesoftware/wildfire/pubsub/DefaultNodeConfiguration.java

@@ -448,13 +448,6 @@ public class DefaultNodeConfiguration {
         formField.setLabel(LocaleUtils.getLocalizedString("pubsub.form.conf.deliver_payloads"));
         formField.addValue(deliverPayloads);
 
-        formField = form.addField();
-        formField.setVariable("pubsub#send_item_subscribe");
-        formField.setType(FormField.Type.boolean_type);
-        formField.setLabel(
-                LocaleUtils.getLocalizedString("pubsub.form.conf.send_item_subscribe"));
-        formField.addValue(sendItemSubscribe);
-
         formField = form.addField();
         formField.setVariable("pubsub#notify_config");
         formField.setType(FormField.Type.boolean_type);
@@ -480,6 +473,13 @@ public class DefaultNodeConfiguration {
         formField.addValue(presenceBasedDelivery);
 
         if (leaf) {
+            formField = form.addField();
+            formField.setVariable("pubsub#send_item_subscribe");
+            formField.setType(FormField.Type.boolean_type);
+            formField.setLabel(
+                    LocaleUtils.getLocalizedString("pubsub.form.conf.send_item_subscribe"));
+            formField.addValue(sendItemSubscribe);
+
             formField = form.addField();
             formField.setVariable("pubsub#persist_items");
             formField.setType(FormField.Type.boolean_type);

src/java/org/jivesoftware/wildfire/pubsub/LeafNode.java

@@ -103,7 +103,7 @@ public class LeafNode extends Node {
             }
         }
         // Remove stored published items based on the new max items
-        while (!publishedItems.isEmpty() && maxPublishedItems > publishedItems.size()) {
+        while (!publishedItems.isEmpty() && publishedItems.size() > maxPublishedItems) {
             PublishedItem removedItem = publishedItems.remove(0);
             itemsByID.remove(removedItem.getID());
             // Add the removed item to the queue of items to delete from the database. The
@@ -236,7 +236,8 @@ public class LeafNode extends Node {
 
                     // Add the published item to the list of items to persist (using another thread)
                     // but check that we don't exceed the limit. Remove oldest items if required.
-                    while (!publishedItems.isEmpty() && maxPublishedItems >= publishedItems.size()) {
+                    while (!publishedItems.isEmpty() && publishedItems.size() >= maxPublishedItems)
+                    {
                         PublishedItem removedItem = publishedItems.remove(0);
                         itemsByID.remove(removedItem.getID());
                         // Add the removed item to the queue of items to delete from the database. The
@@ -410,7 +411,7 @@ public class LeafNode extends Node {
             Element items = event.addElement("purge");
             items.addAttribute("node", nodeID);
             // Send notification that the node configuration has changed
-            broadcastSubscribers(message, false);
+            broadcastNodeEvent(message, false);
         }
     }
 }

src/java/org/jivesoftware/wildfire/pubsub/NodeAffiliate.java

@@ -11,9 +11,9 @@
 
 package org.jivesoftware.wildfire.pubsub;
 
+import org.dom4j.Element;
 import org.xmpp.packet.JID;
 import org.xmpp.packet.Message;
-import org.dom4j.Element;
 
 import java.util.*;
 
@@ -86,7 +86,7 @@ public class NodeAffiliate {
             for (List<NodeSubscription> nodeSubscriptions : itemsBySubs.keySet()) {
                 // Add items information
                 Element items = event.addElement("items");
-                items.addAttribute("node", node.getNodeID());
+                items.addAttribute("node", getNode().getNodeID());
                 for (PublishedItem publishedItem : itemsBySubs.get(nodeSubscriptions)) {
                     // Add item information to the event notification
                     Element item = items.addElement("item");
@@ -96,6 +96,11 @@ public class NodeAffiliate {
                     if (node.isPayloadDelivered()) {
                         item.add(publishedItem.getPayload().createCopy());
                     }
+                    // Add leaf node information if affiliated node and node
+                    // where the item was published are different
+                    if (node != getNode()) {
+                        item.addAttribute("node", node.getNodeID());
+                    }
                 }
                 // Send the event notification
                 sendEventNotification(notification, node, nodeSubscriptions);

src/java/org/jivesoftware/wildfire/pubsub/NodeSubscription.java

@@ -12,14 +12,14 @@
 package org.jivesoftware.wildfire.pubsub;
 
 import org.dom4j.Element;
+import org.jivesoftware.util.FastDateFormat;
 import org.jivesoftware.util.LocaleUtils;
 import org.jivesoftware.util.Log;
-import org.jivesoftware.util.FastDateFormat;
 import org.xmpp.forms.DataForm;
 import org.xmpp.forms.FormField;
+import org.xmpp.packet.IQ;
 import org.xmpp.packet.JID;
 import org.xmpp.packet.Message;
-import org.xmpp.packet.IQ;
 import org.xmpp.packet.Presence;
 
 import java.text.ParseException;
@@ -76,11 +76,6 @@ public class NodeSubscription {
      * Current subscription state.
      */
     private State state;
-    /**
-     * Flag that indicates if configuration is required by the node and is still pending to
-     * be configured by the subscriber.
-     */
-    private boolean configurationPending = false;
     /**
      * Flag indicating whether an entity wants to receive or has disabled notifications.
      */
@@ -130,9 +125,7 @@ public class NodeSubscription {
      */
     private boolean savedToDB = false;
 
-    // TODO Do not send event notifications (e.g. node purge, node deleted) to unconfigured subscriptions ????
-    // TODO Send last published item when a subscription is authorized. We may need to move this logic to another place
-    // TODO Implement presence subscription from the node to the subscriber to figure out if event notifications can be sent 
+    // TODO Implement presence subscription from the node to the subscriber to figure out if event notifications can be sent
 
     static {
         dateFormat = new SimpleDateFormat("yyyy-MM-DD'T'HH:mm:ss.SSS'Z'");
@@ -158,11 +151,6 @@ public class NodeSubscription {
         this.owner = owner;
         this.state = state;
         this.id = id;
-
-        if (node.isSubscriptionConfigurationRequired()) {
-            // Subscription configuration is required and it's still pending
-            setConfigurationPending(true);
-        }
     }
 
     /**
@@ -224,23 +212,25 @@ public class NodeSubscription {
 
     /**
      * Returns true if configuration is required by the node and is still pending to
-     * be configured by the subscriber. Otherwise return false.
+     * be configured by the subscriber. Otherwise return false. Once a subscription is
+     * configured it might need to be approved by a node owner to become active.
      *
      * @return true if configuration is required by the node and is still pending to
      *         be configured by the subscriber.
      */
     public boolean isConfigurationPending() {
-        return configurationPending;
+        return state == State.unconfigured;
     }
 
     /**
-     * Returns true if the subscription was approved by a node owner. Nodes that don't
-     * require node owners to approve subscription assume that all subscriptions are approved.
+     * Returns true if the subscription needs to be approved by a node owner to become
+     * active. Until the subscription is not activated the subscriber will not receive
+     * event notifications.
      *
-     * @return true if the subscription was approved by a node owner.
+     * @return true if the subscription needs to be approved by a node owner to become active.
      */
-    public boolean isApproved() {
-        return State.subscribed == state;
+    public boolean isAuthorizationPending() {
+        return state == State.pending;
     }
 
     /**
@@ -340,10 +330,6 @@ public class NodeSubscription {
         return keyword;
     }
 
-    void setConfigurationPending(boolean configurationPending) {
-        this.configurationPending = configurationPending;
-    }
-
     void setShouldDeliverNotifications(boolean deliverNotifications) {
         this.deliverNotifications = deliverNotifications;
     }
@@ -405,14 +391,22 @@ public class NodeSubscription {
             // Return success response
             service.send(IQ.createResultIQ(originalIQ));
         }
-        // Send last published item if subscription is now configured (and authorized)
-        if (wasUnconfigured && !isConfigurationPending() && node.isSendItemSubscribe()) {
-            PublishedItem lastItem = node.getLastPublishedItem();
-            if (lastItem != null) {
-                sendLastPublishedItem(lastItem);
+
+        if (wasUnconfigured) {
+            // If subscription is pending then send notification to node owners
+            // asking to approve the now configured subscription
+            if (isAuthorizationPending()) {
+                sendAuthorizationRequest();
             }
-        }
 
+            // Send last published item (if node is leaf node and subscription status is ok)
+            if (node.isSendItemSubscribe() && isActive()) {
+                PublishedItem lastItem = node.getLastPublishedItem();
+                if (lastItem != null) {
+                    sendLastPublishedItem(lastItem);
+                }
+            }
+        }
     }
 
     void configure(DataForm options) {
@@ -488,8 +482,13 @@ public class NodeSubscription {
                 fieldExists = false;
             }
             if (fieldExists) {
-                // Mark that the subscription has been configured
-                setConfigurationPending(false);
+                // Subscription has been configured so set the next state
+                if (node.getAccessModel().isAuthorizationRequired()) {
+                    state = State.pending;
+                }
+                else {
+                    state = State.subscribed;
+                }
             }
         }
         if (savedToDB) {
@@ -640,6 +639,32 @@ public class NodeSubscription {
         return true;
     }
 
+    /**
+     * Returns true if node events such as configuration changed or node purged can be
+     * sent to the subscriber.
+     *
+     * @return true if node events such as configuration changed or node purged can be
+     *         sent to the subscriber.
+     */
+    boolean canSendNodeEvents() {
+        // Check if the subscription is active
+        if (!isActive()) {
+            return false;
+        }
+        // Check if delivery of notifications is disabled
+        if (!shouldDeliverNotifications()) {
+            return false;
+        }
+        // Check if delivery is subject to presence-based policy
+        if (!getPresenceStates().isEmpty()) {
+            String show = service.getShowPresence(jid);
+            if (show == null || !getPresenceStates().contains(show)) {
+                return false;
+            }
+        }
+        return true;
+    }
+
     /**
      * Returns true if the published item matches the keyword filter specified in
      * the subscription. If no keyword was specified then answer true.
@@ -662,9 +687,9 @@ public class NodeSubscription {
      *
      * @return true if the subscription is active.
      */
-    private boolean isActive() {
+    public boolean isActive() {
         // Check if subscription is approved and configured (if required)
-        if (!isApproved() || this.isConfigurationPending()) {
+        if (state != State.subscribed) {
             return false;
         }
         // Check if the subscription has expired
@@ -808,6 +833,56 @@ public class NodeSubscription {
         return super.toString() + " - JID: " + getJID() + " - State: " + getState().name();
     }
 
+    /**
+     * The subscription has been approved by a node owner. The subscription is now active so
+     * the subscriber is now allowed to get event notifications.
+     */
+    void approved() {
+        if (state == State.subscribed) {
+            // Do nothing
+            return;
+        }
+        state = State.subscribed;
+
+        if (savedToDB) {
+            // Update the subscription in the backend store
+            PubSubPersistenceManager.saveSubscription(service, node, this, false);
+        }
+
+        // Send last published item (if node is leaf node and subscription status is ok)
+        if (node.isSendItemSubscribe() && isActive()) {
+            PublishedItem lastItem = node.getLastPublishedItem();
+            if (lastItem != null) {
+                sendLastPublishedItem(lastItem);
+            }
+        }
+    }
+
+    /**
+     * Sends an request to authorize the pending subscription to the specified owner.
+     *
+     * @param owner the JID of the user that will get the authorization request.
+     */
+    public void sendAuthorizationRequest(JID owner) {
+        Message authRequest = new Message();
+        authRequest.addExtension(node.getAuthRequestForm(this));
+        authRequest.setTo(owner);
+        authRequest.setFrom(service.getAddress());
+        // Send authentication request to node owners
+        service.send(authRequest);
+    }
+
+    /**
+     * Sends an request to authorize the pending subscription to all owners. The first
+     * answer sent by a owner will be processed. Rest of the answers will be discarded.
+     */
+    public void sendAuthorizationRequest() {
+        Message authRequest = new Message();
+        authRequest.addExtension(node.getAuthRequestForm(this));
+        // Send authentication request to node owners
+        service.broadcast(node, authRequest, node.getOwners());
+    }
+
     /**
      * Subscriptions to a node may exist in several states. Delivery of event notifications
      * varies according to the subscription state of the user with the node.

src/java/org/jivesoftware/wildfire/pubsub/PendingSubscriptionsCommand.java

+/**
+ * $Revision: $
+ * $Date: $
+ *
+ * Copyright (C) 2006 Jive Software. All rights reserved.
+ *
+ * This software is published under the terms of the GNU Public License (GPL),
+ * a copy of which is included in this distribution.
+ */
+
+package org.jivesoftware.wildfire.pubsub;
+
+import org.dom4j.Element;
+import org.jivesoftware.util.LocaleUtils;
+import org.jivesoftware.wildfire.commands.AdHocCommand;
+import org.jivesoftware.wildfire.commands.SessionData;
+import org.xmpp.forms.DataForm;
+import org.xmpp.forms.FormField;
+import org.xmpp.packet.JID;
+
+import java.util.Arrays;
+import java.util.List;
+
+/**
+ * Ad-hoc command that sends pending subscriptions to node owners.
+ *
+ * @author Matt Tucker
+ */
+class PendingSubscriptionsCommand extends AdHocCommand {
+
+    private PubSubService service;
+
+    PendingSubscriptionsCommand(PubSubService service) {
+        this.service = service;
+    }
+
+    protected void addStageInformation(SessionData data, Element command) {
+        DataForm form = new DataForm(DataForm.Type.form);
+        form.setTitle(LocaleUtils.getLocalizedString("pubsub.command.pending-subscriptions.title"));
+        form.addInstruction(
+                LocaleUtils.getLocalizedString("pubsub.command.pending-subscriptions.instruction"));
+
+        FormField formField = form.addField();
+        formField.setVariable("pubsub#node");
+        formField.setType(FormField.Type.list_single);
+        formField.setLabel(
+                LocaleUtils.getLocalizedString("pubsub.command.pending-subscriptions.node"));
+        for (Node node : service.getNodes()) {
+            if (!node.isCollectionNode() && node.isAdmin(data.getOwner())) {
+                formField.addOption(null, node.getNodeID());
+            }
+        }
+        // Add the form to the command
+        command.add(form.getElement());
+    }
+
+    public void execute(SessionData data, Element command) {
+        Element note = command.addElement("note");
+        List<String> nodeIDs = data.getData().get("pubsub#node");
+        if (nodeIDs.isEmpty()) {
+            // No nodeID was provided by the requester
+            note.addAttribute("type", "error");
+            note.setText(LocaleUtils.getLocalizedString(
+                    "pubsub.command.pending-subscriptions.error.idrequired"));
+        }
+        else if (nodeIDs.size() > 1) {
+            // More than one nodeID was provided by the requester
+            note.addAttribute("type", "error");
+            note.setText(LocaleUtils.getLocalizedString(
+                    "pubsub.command.pending-subscriptions.error.manyIDs"));
+        }
+        else {
+            Node node = service.getNode(nodeIDs.get(0));
+            if (node != null) {
+                if (node.isAdmin(data.getOwner())) {
+                    note.addAttribute("type", "info");
+                    note.setText(LocaleUtils.getLocalizedString(
+                            "pubsub.command.pending-subscriptions.success"));
+
+                    for (NodeSubscription subscription : node.getPendingSubscriptions()) {
+                        subscription.sendAuthorizationRequest(data.getOwner());
+                    }
+                }
+                else {
+                    // Requester is not an admin of the specified node
+                    note.addAttribute("type", "error");
+                    note.setText(LocaleUtils.getLocalizedString(
+                            "pubsub.command.pending-subscriptions.error.forbidden"));
+                }
+            }
+            else {
+                // Node with the specified nodeID was not found
+                note.addAttribute("type", "error");
+                note.setText(LocaleUtils.getLocalizedString(
+                        "pubsub.command.pending-subscriptions.error.badid"));
+            }
+        }
+    }
+
+    public String getCode() {
+        return "http://jabber.org/protocol/pubsub#get-pending";
+    }
+
+    public String getDefaultLabel() {
+        return LocaleUtils.getLocalizedString("pubsub.command.pending-subscriptions.label");
+    }
+
+    protected List<Action> getActions(SessionData data) {
+        return Arrays.asList(Action.complete);
+    }
+
+    protected Action getExecuteAction(SessionData data) {
+        return Action.complete;
+    }
+
+    public int getMaxStages(SessionData data) {
+        return 1;
+    }
+
+    public boolean hasPermission(JID requester) {
+        // User has permission if he is an owner of at least one node or is a sysadmin
+        for (Node node : service.getNodes()) {
+            if (!node.isCollectionNode() && node.isAdmin(requester)) {
+                return true;
+            }
+        }
+        return false;
+    }
+}

src/java/org/jivesoftware/wildfire/pubsub/PubSubEngine.java

@@ -18,6 +18,7 @@ import org.jivesoftware.util.LocaleUtils;
 import org.jivesoftware.util.Log;
 import org.jivesoftware.util.StringUtils;
 import org.jivesoftware.wildfire.PacketRouter;
+import org.jivesoftware.wildfire.commands.AdHocCommandManager;
 import org.jivesoftware.wildfire.pubsub.models.AccessModel;
 import org.jivesoftware.wildfire.user.UserManager;
 import org.xmpp.forms.DataForm;
@@ -35,6 +36,10 @@ import java.util.concurrent.LinkedBlockingQueue;
 public class PubSubEngine {
 
     private PubSubService service;
+    /**
+     * Manager that keeps the list of ad-hoc commands and processing command requests.
+     */
+    private AdHocCommandManager manager;
     /**
      * The time to elapse between each execution of the maintenance process. Default
      * is 2 minutes.
@@ -70,7 +75,9 @@ public class PubSubEngine {
     public PubSubEngine(PubSubService pubSubService, PacketRouter router) {
         this.service = pubSubService;
         this.router = router;
-
+        // Initialize the ad-hoc commands manager to use for this pubsub service
+        manager = new AdHocCommandManager();
+        manager.addCommand(new PendingSubscriptionsCommand(service));
         // Save or delete published items from the database every 2 minutes starting in
         // 2 minutes (default values)
         publishedItemTask = new PublishedItemTask();
@@ -211,6 +218,11 @@ public class PubSubEngine {
             sendErrorPacket(iq, PacketError.Condition.bad_request, null);
             return true;
         }
+        else if ("http://jabber.org/protocol/commands".equals(namespace)) {
+            // Process ad-hoc command
+            IQ reply = manager.process(iq);
+            router.route(reply);
+        }
         return false;
     }
 
@@ -224,13 +236,53 @@ public class PubSubEngine {
     }
 
     /**
-     * Handles Message packets sent to the pubsub service.
+     * Handles Message packets sent to the pubsub service. Messages may be of type error
+     * when an event notification was sent to a susbcriber whose address is no longer available.<p>
+     *
+     * Answers to authorization requests sent to node owners to approve pending subscriptions
+     * will also be processed by this method.
      *
      * @param message the Message packet sent to the pubsub service.
      */
     public void process(Message message) {
         // TODO Process Messages of type error to identify possible subscribers that no longer exist
-        // See "Handling Notification-Related Errors" section
+        if (message.getType() == Message.Type.error) {
+            // See "Handling Notification-Related Errors" section
+        }
+        else if (message.getType() == Message.Type.normal) {
+            // Check that this is an answer to an authorization request
+            DataForm authForm = (DataForm) message.getExtension("x", "jabber:x:data");
+            if (authForm != null && authForm.getType() == DataForm.Type.submit) {
+                String formType = authForm.getField("FORM_TYPE").getValues().get(0);
+                // Check that completed data form belongs to an authorization request
+                if ("http://jabber.org/protocol/pubsub#subscribe_authorization".equals(formType)) {
+                    String nodeID = authForm.getField("pubsub#node").getValues().get(0);
+                    String subID = authForm.getField("pubsub#subid").getValues().get(0);
+                    String allow = authForm.getField("pubsub#allow").getValues().get(0);
+                    boolean approved;
+                    if ("1".equals(allow) || "true".equals(allow)) {
+                        approved = true;
+                    }
+                    else if ("0".equals(allow) || "false".equals(allow)) {
+                        approved = false;
+                    }
+                    else {
+                        // Unknown allow value. Ignore completed form
+                        Log.warn("Invalid allow value in completed authorization form: " +
+                                message.toXML());
+                        return;
+                    }
+                    // Approve or cancel the pending subscription to the node
+                    Node node = service.getNode(nodeID);
+                    if (node != null) {
+                        NodeSubscription subscription = node.getSubscription(subID);
+                        if (subscription != null) {
+                            node.approveSubscription(subscription, approved);
+                        }
+                    }
+                }
+            }
+        }
     }
 
     private void publishItemsToNode(IQ iq, Element publishElement) {
@@ -1322,6 +1374,8 @@ public class PubSubEngine {
                 PubSubPersistenceManager.createPublishedItem(service, entry);
             }
         }
+        // Stop executing ad-hoc commands
+        manager.stop();
     }
 
     /*******************************************************************************

src/java/org/jivesoftware/wildfire/pubsub/PubSubService.java

@@ -34,6 +34,13 @@ import java.util.Collection;
  */
 public interface PubSubService {
 
+    /**
+     * Returns the XMPP address of the service.
+     *
+     * @return the XMPP address of the service.
+     */
+    JID getAddress();
+
     /**
      * Returns a String that uniquely identifies this pubsub service. This information is
      * being used when storing node information in the database so it's possible to have

src/java/org/jivesoftware/wildfire/pubsub/models/AuthorizeAccess.java

@@ -11,14 +11,14 @@
 
 package org.jivesoftware.wildfire.pubsub.models;
 
+import org.dom4j.DocumentHelper;
+import org.dom4j.Element;
+import org.dom4j.QName;
 import org.jivesoftware.wildfire.pubsub.Node;
-import org.jivesoftware.wildfire.pubsub.NodeSubscription;
 import org.jivesoftware.wildfire.pubsub.NodeAffiliate;
+import org.jivesoftware.wildfire.pubsub.NodeSubscription;
 import org.xmpp.packet.JID;
 import org.xmpp.packet.PacketError;
-import org.dom4j.Element;
-import org.dom4j.DocumentHelper;
-import org.dom4j.QName;
 
 /**
  * Subscription requests must be approved and only subscribers may retrieve items.
@@ -43,7 +43,7 @@ public class AuthorizeAccess extends AccessModel {
         // Any subscription of this entity that was approved will give him access
         // to retrieve the node items
         for (NodeSubscription subscription : nodeAffiliate.getSubscriptions()) {
-            if (subscription.isApproved()) {
+            if (subscription.isActive()) {
                 return true;
             }
         }

src/java/org/jivesoftware/wildfire/pubsub/models/OnlySubscribers.java

@@ -37,7 +37,7 @@ public class OnlySubscribers extends PublisherModel {
         }
         // Grant access if at least one subscription of this user was approved
         for (NodeSubscription subscription : nodeAffiliate.getSubscriptions()) {
-            if (subscription.isApproved()) {
+            if (subscription.isActive()) {
                 return true;
             }
         }