Intergation day...untested code.

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

src/java/org/jivesoftware/wildfire/commands/admin/PacketsNotification.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.commands.admin;
+
+import org.dom4j.Element;
+import org.jivesoftware.wildfire.commands.AdHocCommand;
+import org.jivesoftware.wildfire.commands.SessionData;
+import org.jivesoftware.wildfire.component.InternalComponentManager;
+import org.jivesoftware.wildfire.interceptor.PacketNotificator;
+import org.xmpp.component.Component;
+import org.xmpp.forms.DataForm;
+import org.xmpp.forms.FormField;
+import org.xmpp.packet.JID;
+
+import java.util.Arrays;
+import java.util.List;
+
+/**
+ * Command that allows to retrieve the presence of all active users.
+ *
+ * @author Gaston Dombiak
+ *
+ * TODO Use i18n
+ * TODO Create command for removing subscriptions. Subscriptions will now be removed when component disconnects.
+ */
+public class PacketsNotification extends AdHocCommand {
+
+    protected void addStageInformation(SessionData data, Element command) {
+        DataForm form = new DataForm(DataForm.Type.form);
+        form.setTitle("Receiving notification of packets activity");
+        form.addInstruction("Fill out this form to configure packets to receive.");
+
+        FormField field = form.addField();
+        field.setType(FormField.Type.hidden);
+        field.setVariable("FORM_TYPE");
+        field.addValue("http://jabber.org/protocol/admin");
+
+        field = form.addField();
+        field.setType(FormField.Type.list_multi);
+        field.setLabel("Type of packet");
+        field.setVariable("packet_type");
+        field.addOption("Presence", "presence");
+        field.addOption("IQ", "iq");
+        field.addOption("Message", "message");
+        field.setRequired(true);
+
+        field = form.addField();
+        field.setType(FormField.Type.list_single);
+        field.setLabel("Direction");
+        field.setVariable("direction");
+        field.addOption("Incoming", "incoming");
+        field.addOption("Outgoing", "outgoing");
+        field.setRequired(true);
+
+        field = form.addField();
+        field.setType(FormField.Type.list_single);
+        field.setLabel("Processing time");
+        field.setVariable("processed");
+        field.addOption("Before processing", "false");
+        field.addOption("After processing", "true");
+        field.setRequired(true);
+
+        // Add the form to the command
+        command.add(form.getElement());
+    }
+
+    public void execute(SessionData data, Element command) {
+        boolean presenceEnabled = false;
+        boolean messageEnabled = false;
+        boolean iqEnabled = false;
+        for (String packet_type : data.getData().get("packet_type")) {
+            if ("presence".equals(packet_type)) {
+                presenceEnabled = true;
+            }
+            else if ("iq".equals(packet_type)) {
+                iqEnabled = true;
+            }
+            else if ("message".equals(packet_type)) {
+                messageEnabled = true;
+            }
+        }
+
+        boolean incoming = "incoming".equals(data.getData().get("direction").get(0));
+        boolean processed = "true".equals(data.getData().get("processed").get(0));
+
+        JID componentJID = data.getOwner();
+        Component component = InternalComponentManager.getInstance().getComponent(componentJID);
+        // Create or update subscription of the component to receive packet notifications
+        PacketNotificator.getInstance().addSubscriber(componentJID, component, iqEnabled,
+                messageEnabled, presenceEnabled, incoming, processed);
+
+        // Inform that everything went fine
+        Element note = command.addElement("note");
+        note.addAttribute("type", "info");
+        note.setText("Operation finished successfully");
+    }
+
+    public String getCode() {
+        return "http://jabber.org/protocol/admin#packets_notification";
+    }
+
+    public String getDefaultLabel() {
+        return "Get notifications of packet activity";
+    }
+
+    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;
+    }
+
+    /**
+     * Returns if the requester can access this command. Only components are allowed to
+     * execute this command.
+     *
+     * @param requester the JID of the user requesting to execute this command.
+     * @return true if the requester can access this command.
+     */
+    public boolean hasPermission(JID requester) {
+        return InternalComponentManager.getInstance().getComponent(requester) != null;
+    }
+}

src/java/org/jivesoftware/wildfire/interceptor/PacketNotificator.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.interceptor;
+
+import org.dom4j.Element;
+import org.jivesoftware.util.FastDateFormat;
+import org.jivesoftware.util.JiveConstants;
+import org.jivesoftware.util.LocaleUtils;
+import org.jivesoftware.util.Log;
+import org.jivesoftware.wildfire.Session;
+import org.jivesoftware.wildfire.XMPPServer;
+import org.jivesoftware.wildfire.component.ComponentEventListener;
+import org.jivesoftware.wildfire.component.InternalComponentManager;
+import org.xmpp.component.Component;
+import org.xmpp.packet.*;
+
+import java.util.*;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.LinkedBlockingQueue;
+
+/**
+ * Packet interceptor that notifies of packets activity to components that previously
+ * subscribed to the notificator. Notifications to components will be made using
+ * a Message sent to the component itself. The Message will include an extension that will
+ * contain the intercepted packet as well as extra information such us <tt>incoming</tt>
+ * and <tt>processed</tt>.
+ *
+ * @author Gaston Dombiak
+ */
+public class PacketNotificator implements PacketInterceptor, ComponentEventListener {
+
+    private final static PacketNotificator instance = new PacketNotificator();
+    private final static FastDateFormat dateFormat = FastDateFormat
+            .getInstance(JiveConstants.XMPP_DELAY_DATETIME_FORMAT, TimeZone.getTimeZone("UTC"));
+
+    private Map<String, Subscription> subscribers = new ConcurrentHashMap<String, Subscription>();
+    private String serverName;
+
+    /**
+     * Timer to save queued logs to the XML file.
+     */
+    private Timer timer = new Timer("PacketActivityNotifier");
+    private ProcessPacketsTask packetsTask;
+
+    /**
+     * Queue that holds the audited packets that will be later saved to an XML file.
+     */
+    private BlockingQueue<InterceptedPacket> packetQueue =
+            new LinkedBlockingQueue<InterceptedPacket>();
+
+    /**
+     * Returns unique instance of this class.
+     *
+     * @return unique instance of this class.
+     */
+    public static PacketNotificator getInstance() {
+        return instance;
+    }
+
+
+    private PacketNotificator() {
+        // Add the new instance as a listener of component events. We need to react when
+        // a component is no longer valid
+        InternalComponentManager.getInstance().addListener(this);
+        serverName = XMPPServer.getInstance().getServerInfo().getName();
+
+        // Create a new task and schedule it with the new timeout
+        packetsTask = new ProcessPacketsTask();
+        timer.schedule(packetsTask, 5000, 5000);
+    }
+
+    /**
+     * Creates new subscription for the specified component with the specified settings.
+     *
+     * @param componentJID the address of the component connected to the server.
+     * @param component the component that will be notified of packet activity.
+     * @param iqEnabled true if interested in IQ packets of any type.
+     * @param messageEnabled true if interested in Message packets.
+     * @param presenceEnabled true if interested in Presence packets.
+     * @param incoming true if interested in incoming traffic. false means outgoing.
+     * @param processed true if want to be notified after packets were processed.
+     */
+    public void addSubscriber(JID componentJID, Component component, boolean iqEnabled,
+            boolean messageEnabled, boolean presenceEnabled, boolean incoming, boolean processed) {
+        subscribers.put(componentJID.toString(), new Subscription(component, iqEnabled,
+                messageEnabled, presenceEnabled, incoming, processed));
+    }
+
+    /**
+     * Removes the subscription of the specified component.
+     *
+     * @param componentJID the address of the component connected to the server.
+     */
+    public void removeSubscriber(JID componentJID) {
+        subscribers.remove(componentJID.toString());
+    }
+
+    public void interceptPacket(Packet packet, Session session, boolean incoming, boolean processed)
+            throws PacketRejectedException {
+        // Queue intercepted packet only if there are subscribers interested
+        if (!subscribers.isEmpty()) {
+            boolean queue = false;
+            Class packetClass = packet.getClass();
+            for (Subscription subscription : subscribers.values()) {
+                if (subscription.isPresenceEnabled() && packetClass == Presence.class) {
+                    queue = true;
+                }
+                else if (subscription.isMessageEnabled() && packetClass == Message.class) {
+                    queue = true;
+                }
+                else if (subscription.isIQEnabled() && packetClass == IQ.class) {
+                    queue = true;
+                }
+            }
+            if (queue) {
+                // Queue packet with extra information and let the background thread process it
+                packetQueue.add(new InterceptedPacket(packet, incoming, processed));
+            }
+        }
+    }
+
+    public void componentInfoReceived(Component component, IQ iq) {
+        //Ignore
+    }
+
+    public void componentRegistered(Component component, JID componentJID) {
+        //Ignore
+    }
+
+    public void componentUnregistered(Component component, JID componentJID) {
+        //Remove component from the list of subscribers (if subscribed)
+        removeSubscriber(componentJID);
+    }
+
+    private void processPackets() {
+        List<InterceptedPacket> packets = new ArrayList<InterceptedPacket>(packetQueue.size());
+        packetQueue.drainTo(packets);
+        for (InterceptedPacket interceptedPacket : packets) {
+            for (Map.Entry<String, Subscription> entry : subscribers.entrySet()) {
+                boolean notify = false;
+                String componentJID = entry.getKey();
+                Subscription subscription = entry.getValue();
+
+                if (subscription.isIncoming() == interceptedPacket.isIncoming() &&
+                        subscription.isProcessed() == interceptedPacket.isProcessed()) {
+                    Class packetClass = interceptedPacket.getPacketClass();
+                    if (subscription.isPresenceEnabled() && packetClass == Presence.class) {
+                        notify = true;
+                    }
+                    else if (subscription.isMessageEnabled() && packetClass == Message.class) {
+                        notify = true;
+                    }
+                    else if (subscription.isIQEnabled() && packetClass == IQ.class) {
+                        notify = true;
+                    }
+                }
+
+                if (notify) {
+                    try {
+                        Message message = new Message();
+                        message.setFrom(serverName);
+                        message.setTo(componentJID);
+                        Element childElement = message.addChildElement("copy",
+                                "http://jabber.org/protocol/packet#event");
+                        childElement.addAttribute("incoming", subscription.isIncoming() ? "true" : "false");
+                        childElement.addAttribute("processed", subscription.isProcessed() ? "true" : "false");
+                        childElement.addAttribute("date", dateFormat.format(interceptedPacket.getCreationDate()));
+                        childElement.add(interceptedPacket.getElement().createCopy());
+                    }
+                    catch (Exception e) {
+                        Log.error(LocaleUtils.getLocalizedString("admin.error"), e);
+                    }
+                }
+            }
+        }
+    }
+
+    private class ProcessPacketsTask extends TimerTask {
+        public void run() {
+            try {
+                // Notify components of intercepted packets
+                processPackets();
+            }
+            catch (Throwable e) {
+                Log.error(LocaleUtils.getLocalizedString("admin.error"), e);
+            }
+        }
+    }
+
+    private static class Subscription {
+        private Component component;
+        private boolean presenceEnabled;
+        private boolean messageEnabled;
+        private boolean iqEnabled;
+        private boolean incoming;
+        private boolean processed;
+
+        /**
+         * Creates a new subscription for the specified Component with the
+         * specified configuration.
+         *
+         * @param component the component subscribing.
+         * @param iqEnabled true if interested in IQ packets of any type.
+         * @param messageEnabled true if interested in Message packets.
+         * @param presenceEnabled true if interested in Presence packets.
+         * @param incoming true if interested in incoming traffic. false means outgoing.
+         * @param processed true if want to be notified after packets were processed.
+         */
+        public Subscription(Component component, boolean iqEnabled, boolean messageEnabled,
+                boolean presenceEnabled, boolean incoming, boolean processed) {
+            this.component = component;
+            this.incoming = incoming;
+            this.iqEnabled = iqEnabled;
+            this.messageEnabled = messageEnabled;
+            this.presenceEnabled = presenceEnabled;
+            this.processed = processed;
+        }
+
+        /**
+         * Returns the component that is subscribed.
+         *
+         * @return the component that is subscribed.
+         */
+        public Component getComponent() {
+            return component;
+        }
+
+        /**
+         * Returns true if the component is interested in receiving notifications
+         * of intercepted IQ packets.
+         *
+         * @return true if the component is interested in receiving notifications
+         * of intercepted IQ packets.
+         */
+        public boolean isIQEnabled() {
+            return iqEnabled;
+        }
+
+        /**
+         * Returns true if the component is interested in receiving notifications
+         * of intercepted Message packets.
+         *
+         * @return true if the component is interested in receiving notifications
+         * of intercepted Message packets.
+         */
+        public boolean isMessageEnabled() {
+            return messageEnabled;
+        }
+
+        /**
+         * Returns true if the component is interested in receiving notifications
+         * of intercepted Presence packets.
+         *
+         * @return true if the component is interested in receiving notifications
+         * of intercepted Presence packets.
+         */
+        public boolean isPresenceEnabled() {
+            return presenceEnabled;
+        }
+
+        /**
+         * Returns true if the component wants to be notified of incoming traffic. A false
+         * value means that the component is interested in outgoing traffic.
+         *
+         * @return true if interested in incoming traffic. false means outgoing.
+         */
+        public boolean isIncoming() {
+            return incoming;
+        }
+
+        /**
+         * Returns true if the component wants to be notified of after packets were
+         * processed. Processed has different meaning depending if the packet is incoming
+         * or outgoing. An incoming packet that was processed means that the server
+         * routed the packet to the recipient and nothing else should be done with the packet.
+         * However, an outgoing packet that was processed means that the packet was already
+         * sent to the target entity. 
+         *
+         * @return true if interested in incoming traffic. false means outgoing.
+         */
+        public boolean isProcessed() {
+            return processed;
+        }
+    }
+
+    private static class InterceptedPacket {
+        private Element element;
+        private Class packetClass;
+        private Date creationDate;
+        private boolean incoming;
+        private boolean processed;
+
+        public InterceptedPacket(Packet packet, boolean incoming, boolean processed) {
+            packetClass = packet.getClass();
+            this.element = packet.getElement();
+            this.incoming = incoming;
+            this.processed = processed;
+            creationDate = new Date();
+        }
+
+
+        public Class getPacketClass() {
+            return packetClass;
+        }
+
+        public Date getCreationDate() {
+            return creationDate;
+        }
+
+        public Element getElement() {
+            return element;
+        }
+
+        public boolean isIncoming() {
+            return incoming;
+        }
+
+        public boolean isProcessed() {
+            return processed;
+        }
+    }
+}