Work on components in a clustered environment.

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

src/java/org/jivesoftware/openfire/XMPPServer.java

@@ -193,7 +193,7 @@ public class XMPPServer {
      */
     public boolean isRemote(JID jid) {
         if (jid != null) {
-            if (!name.equals(jid.getDomain()) && componentManager.getComponent(jid) == null) {
+            if (!name.equals(jid.getDomain()) && !componentManager.hasComponent(jid)) {
                 return true;
             }
         }
@@ -229,8 +229,7 @@ public class XMPPServer {
      * @return true if the given address matches a component service JID.
      */
     public boolean matchesComponent(JID jid) {
-        return jid != null && !name.equals(jid.getDomain()) &&
-                componentManager.getComponent(jid) != null;
+        return jid != null && !name.equals(jid.getDomain()) && componentManager.hasComponent(jid);
     }
 
     /**

src/java/org/jivesoftware/openfire/commands/admin/GetServerStats.java

@@ -12,15 +12,15 @@ package org.jivesoftware.openfire.commands.admin;
 
 import org.dom4j.Element;
 import org.jivesoftware.admin.AdminConsole;
-import org.jivesoftware.util.FastDateFormat;
-import org.jivesoftware.util.JiveConstants;
-import org.jivesoftware.util.LocaleUtils;
 import org.jivesoftware.openfire.SessionManager;
 import org.jivesoftware.openfire.XMPPServer;
 import org.jivesoftware.openfire.commands.AdHocCommand;
 import org.jivesoftware.openfire.commands.SessionData;
 import org.jivesoftware.openfire.component.InternalComponentManager;
 import org.jivesoftware.openfire.session.ClientSession;
+import org.jivesoftware.util.FastDateFormat;
+import org.jivesoftware.util.JiveConstants;
+import org.jivesoftware.util.LocaleUtils;
 import org.xmpp.forms.DataForm;
 import org.xmpp.forms.FormField;
 import org.xmpp.packet.JID;
@@ -151,6 +151,6 @@ public class GetServerStats extends AdHocCommand {
      * @return true if the requester can access this command.
      */
     public boolean hasPermission(JID requester) {
-        return super.hasPermission(requester) || InternalComponentManager.getInstance().getComponent(requester) != null;
+        return super.hasPermission(requester) || InternalComponentManager.getInstance().hasComponent(requester);
     }
 }

src/java/org/jivesoftware/openfire/commands/admin/GetUsersPresence.java

@@ -124,7 +124,6 @@ public class GetUsersPresence extends AdHocCommand {
      * @return true if the requester can access this command.
      */
     public boolean hasPermission(JID requester) {
-        return super.hasPermission(requester) ||
-                InternalComponentManager.getInstance().getComponent(requester) != null;
+        return super.hasPermission(requester) || InternalComponentManager.getInstance().hasComponent(requester);
     }
 }

src/java/org/jivesoftware/openfire/commands/admin/HttpBindStatus.java

@@ -8,17 +8,17 @@
  */
 package org.jivesoftware.openfire.commands.admin;
 
+import org.dom4j.Element;
 import org.jivesoftware.openfire.commands.AdHocCommand;
 import org.jivesoftware.openfire.commands.SessionData;
-import org.jivesoftware.openfire.http.HttpBindManager;
 import org.jivesoftware.openfire.component.InternalComponentManager;
-import org.dom4j.Element;
+import org.jivesoftware.openfire.http.HttpBindManager;
 import org.xmpp.forms.DataForm;
 import org.xmpp.forms.FormField;
 import org.xmpp.packet.JID;
 
-import java.util.List;
 import java.util.Collections;
+import java.util.List;
 
 /**
  * AdHoc command to return the current status of the HTTP-bind service. The command returns whether
@@ -93,7 +93,6 @@ public class HttpBindStatus extends AdHocCommand {
 
     @Override
     public boolean hasPermission(JID requester) {
-        return super.hasPermission(requester) ||
-                InternalComponentManager.getInstance().getComponent(requester) != null;
+        return super.hasPermission(requester) || InternalComponentManager.getInstance().hasComponent(requester);
     }
 }

src/java/org/jivesoftware/openfire/commands/admin/PacketsNotification.java

@@ -15,7 +15,6 @@ import org.jivesoftware.openfire.commands.AdHocCommand;
 import org.jivesoftware.openfire.commands.SessionData;
 import org.jivesoftware.openfire.component.InternalComponentManager;
 import org.jivesoftware.openfire.interceptor.PacketCopier;
-import org.xmpp.component.Component;
 import org.xmpp.forms.DataForm;
 import org.xmpp.forms.FormField;
 import org.xmpp.packet.JID;
@@ -92,10 +91,9 @@ public class PacketsNotification extends AdHocCommand {
         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
-        PacketCopier.getInstance().addSubscriber(componentJID, component, iqEnabled,
-                messageEnabled, presenceEnabled, incoming, processed);
+        PacketCopier.getInstance()
+                .addSubscriber(componentJID, iqEnabled, messageEnabled, presenceEnabled, incoming, processed);
 
         // Inform that everything went fine
         Element note = command.addElement("note");
@@ -131,6 +129,6 @@ public class PacketsNotification extends AdHocCommand {
      * @return true if the requester can access this command.
      */
     public boolean hasPermission(JID requester) {
-        return InternalComponentManager.getInstance().getComponent(requester) != null;
+        return InternalComponentManager.getInstance().hasComponent(requester);
     }
 }

src/java/org/jivesoftware/openfire/commands/admin/user/AuthenticateUser.java

@@ -8,22 +8,22 @@
  */
 package org.jivesoftware.openfire.commands.admin.user;
 
+import org.dom4j.Element;
+import org.jivesoftware.openfire.XMPPServer;
+import org.jivesoftware.openfire.auth.AuthFactory;
+import org.jivesoftware.openfire.auth.UnauthorizedException;
 import org.jivesoftware.openfire.commands.AdHocCommand;
 import org.jivesoftware.openfire.commands.SessionData;
-import org.jivesoftware.openfire.user.UserManager;
+import org.jivesoftware.openfire.component.InternalComponentManager;
 import org.jivesoftware.openfire.user.User;
+import org.jivesoftware.openfire.user.UserManager;
 import org.jivesoftware.openfire.user.UserNotFoundException;
-import org.jivesoftware.openfire.auth.AuthFactory;
-import org.jivesoftware.openfire.auth.UnauthorizedException;
-import org.jivesoftware.openfire.XMPPServer;
-import org.jivesoftware.openfire.component.InternalComponentManager;
-import org.dom4j.Element;
 import org.xmpp.forms.DataForm;
 import org.xmpp.forms.FormField;
 import org.xmpp.packet.JID;
 
-import java.util.List;
 import java.util.Arrays;
+import java.util.List;
 
 /**
  * Takes a user's username and password to authenticate them against the Openfire authprovider.
@@ -128,7 +128,6 @@ public class AuthenticateUser extends AdHocCommand {
 
     @Override
     public boolean hasPermission(JID requester) {
-        return super.hasPermission(requester) ||
-                InternalComponentManager.getInstance().getComponent(requester) != null;
+        return super.hasPermission(requester) || InternalComponentManager.getInstance().hasComponent(requester);
     }
 }

src/java/org/jivesoftware/openfire/component/InternalComponentManager.java

@@ -14,6 +14,7 @@ package org.jivesoftware.openfire.component;
 import org.dom4j.Element;
 import org.jivesoftware.openfire.*;
 import org.jivesoftware.openfire.container.BasicModule;
+import org.jivesoftware.openfire.disco.IQDiscoItemsHandler;
 import org.jivesoftware.openfire.session.ComponentSession;
 import org.jivesoftware.util.JiveGlobals;
 import org.jivesoftware.util.Log;
@@ -23,8 +24,6 @@ import org.xmpp.component.ComponentManager;
 import org.xmpp.component.ComponentManagerFactory;
 import org.xmpp.packet.*;
 
-import java.util.Collection;
-import java.util.Collections;
 import java.util.List;
 import java.util.Map;
 import java.util.concurrent.ConcurrentHashMap;
@@ -52,8 +51,7 @@ public class InternalComponentManager extends BasicModule implements ComponentMa
     /**
      * Holds the list of listeners that will be notified of component events.
      */
-    private List<ComponentEventListener> listeners =
-            new CopyOnWriteArrayList<ComponentEventListener>();
+    private List<ComponentEventListener> listeners = new CopyOnWriteArrayList<ComponentEventListener>();
 
     private static InternalComponentManager instance;
     /**
@@ -65,6 +63,7 @@ public class InternalComponentManager extends BasicModule implements ComponentMa
      * in many methods.
      */
     private String serverDomain;
+    private RoutingTable routingTable;
 
     public InternalComponentManager() {
         super("Internal Component Manager");
@@ -75,6 +74,11 @@ public class InternalComponentManager extends BasicModule implements ComponentMa
         return instance;
     }
 
+    public void initialize(XMPPServer server) {
+        super.initialize(server);
+        routingTable = server.getRoutingTable();
+    }
+
     public void start() {
         // Set this ComponentManager as the current component manager
         ComponentManagerFactory.setComponentManager(instance);
@@ -155,13 +159,15 @@ public class InternalComponentManager extends BasicModule implements ComponentMa
         JID componentJID = new JID(subdomain + "." + serverDomain);
 
         // Remove the route for the service provided by the component
-        if (XMPPServer.getInstance().getRoutingTable() != null) {
-            XMPPServer.getInstance().getRoutingTable().removeComponentRoute(componentJID);
+        RoutingTable routingTable = XMPPServer.getInstance().getRoutingTable();
+        if (routingTable != null) {
+            routingTable.removeComponentRoute(componentJID);
         }
 
         // Remove the disco item from the server for the component that is being removed
-        if (XMPPServer.getInstance().getIQDiscoItemsHandler() != null) {
-            XMPPServer.getInstance().getIQDiscoItemsHandler().removeComponentItem(componentJID.toBareJID());
+        IQDiscoItemsHandler iqDiscoItemsHandler = XMPPServer.getInstance().getIQDiscoItemsHandler();
+        if (iqDiscoItemsHandler != null) {
+            iqDiscoItemsHandler.removeComponentItem(componentJID.toBareJID());
         }
 
         // Ask the component to shutdown
@@ -169,7 +175,7 @@ public class InternalComponentManager extends BasicModule implements ComponentMa
             component.shutdown();
         }
 
-        // Notify listeners that a new component has been registered
+        // Notify listeners that an existing component has been unregistered
         for (ComponentEventListener listener : listeners) {
             listener.componentUnregistered(component, componentJID);
         }
@@ -319,23 +325,15 @@ public class InternalComponentManager extends BasicModule implements ComponentMa
     }
 
     /**
-     * Returns the list of components that are currently installed in the server.
-     * This includes internal and external components.
-     *
-     * @return the list of installed components.
-     */
-    public Collection<Component> getComponents() {
-        return Collections.unmodifiableCollection(components.values());
-    }
-
-    /**
-     * Retrieves the <code>Component</code> which is mapped
-     * to the specified JID.
+     * Retrieves the <code>Component</code> which is mapped to the specified JID. The
+     * look up will only be done on components that were registered with this JVM. That
+     * means that components registered in other cluster nodes are not going to be
+     * considered.
      *
      * @param componentJID the jid mapped to the component.
      * @return the component with the specified id.
      */
-    public Component getComponent(JID componentJID) {
+    private Component getComponent(JID componentJID) {
         if (componentJID.getNode() != null) {
             return null;
         }
@@ -356,14 +354,20 @@ public class InternalComponentManager extends BasicModule implements ComponentMa
     }
 
     /**
-     * Retrieves the <code>Component</code> which is mapped
-     * to the specified JID.
+     * Returns true if a component is associated to the specified address. Components
+     * registered with this JVM or other cluster nodes are going to be considered.
      *
-     * @param jid the jid mapped to the component.
-     * @return the component with the specified id.
+     * @param componentJID the address of the component. This is the complete domain.
+     * @return true if a component is associated to the specified address.
      */
-    public Component getComponent(String jid) {
-        return getComponent(new JID(jid));
+    public boolean hasComponent(JID componentJID) {
+        if (componentJID.getNode() != null || componentJID.getResource() != null) {
+            return false;
+        }
+//        if (componentJID.getDomain().lastIndexOf("." + serverDomain) == -1) {
+//            componentJID = new JID(componentJID.getDomain() + "." + serverDomain);
+//        }
+        return routingTable.hasComponentRoute(componentJID);
     }
 
     /**
@@ -380,12 +384,11 @@ public class InternalComponentManager extends BasicModule implements ComponentMa
         for (JID prober : presenceMap.keySet()) {
             JID probee = presenceMap.get(prober);
 
-            Component component = getComponent(probee);
-            if (component != null) {
+            if (routingTable.hasComponentRoute(probee)) {
                 Presence presence = new Presence();
                 presence.setFrom(prober);
                 presence.setTo(probee);
-                component.processPacket(presence);
+                routingTable.routePacket(probee, presence);
 
                 // No reason to hold onto prober reference.
                 presenceMap.remove(prober);

src/java/org/jivesoftware/openfire/disco/IQDiscoItemsHandler.java

@@ -56,8 +56,7 @@ public class IQDiscoItemsHandler extends IQHandler implements ServerFeaturesProv
 
     private Map<String,DiscoItemsProvider> entities = new HashMap<String,DiscoItemsProvider>();
     private List<Element> serverItems = new ArrayList<Element>();
-    private Map<String, DiscoItemsProvider> serverNodeProviders =
-            new ConcurrentHashMap<String, DiscoItemsProvider>();
+    private Map<String, DiscoItemsProvider> serverNodeProviders = new ConcurrentHashMap<String, DiscoItemsProvider>();
     private IQHandlerInfo info;
     private IQDiscoInfoHandler infoHandler;
 

src/java/org/jivesoftware/openfire/interceptor/PacketCopier.java

@@ -11,14 +11,15 @@
 package org.jivesoftware.openfire.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.openfire.RoutingTable;
 import org.jivesoftware.openfire.XMPPServer;
 import org.jivesoftware.openfire.component.ComponentEventListener;
 import org.jivesoftware.openfire.component.InternalComponentManager;
 import org.jivesoftware.openfire.session.Session;
+import org.jivesoftware.util.FastDateFormat;
+import org.jivesoftware.util.JiveConstants;
+import org.jivesoftware.util.LocaleUtils;
+import org.jivesoftware.util.Log;
 import org.xmpp.component.Component;
 import org.xmpp.packet.*;
 
@@ -44,6 +45,7 @@ public class PacketCopier implements PacketInterceptor, ComponentEventListener {
 
     private Map<String, Subscription> subscribers = new ConcurrentHashMap<String, Subscription>();
     private String serverName;
+    private RoutingTable routingTable;
 
     /**
      * Timer to save queued logs to the XML file.
@@ -54,8 +56,7 @@ public class PacketCopier implements PacketInterceptor, ComponentEventListener {
     /**
      * Queue that holds the audited packets that will be later saved to an XML file.
      */
-    private BlockingQueue<InterceptedPacket> packetQueue =
-            new LinkedBlockingQueue<InterceptedPacket>();
+    private BlockingQueue<InterceptedPacket> packetQueue = new LinkedBlockingQueue<InterceptedPacket>();
 
     /**
      * Returns unique instance of this class.
@@ -71,7 +72,9 @@ public class PacketCopier implements PacketInterceptor, ComponentEventListener {
         // 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();
+        XMPPServer server = XMPPServer.getInstance();
+        serverName = server.getServerInfo().getName();
+        routingTable = server.getRoutingTable();
 
         // Add new instance to the PacketInterceptors list
         InterceptorManager.getInstance().addInterceptor(this);
@@ -85,17 +88,16 @@ public class PacketCopier implements PacketInterceptor, ComponentEventListener {
      * 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));
+    public void addSubscriber(JID componentJID, boolean iqEnabled, boolean messageEnabled, boolean presenceEnabled,
+                              boolean incoming, boolean processed) {
+        subscribers.put(componentJID.toString(),
+                new Subscription(iqEnabled, messageEnabled, presenceEnabled, incoming, processed));
     }
 
     /**
@@ -179,7 +181,7 @@ public class PacketCopier implements PacketInterceptor, ComponentEventListener {
                         childElement.addAttribute("date", dateFormat.format(interceptedPacket.getCreationDate()));
                         childElement.add(interceptedPacket.getElement().createCopy());
                         // Send message notification to subscribed component
-                        subscription.getComponent().processPacket(message);
+                        routingTable.routePacket(message.getTo(), message);
                     }
                     catch (Exception e) {
                         Log.error(LocaleUtils.getLocalizedString("admin.error"), e);
@@ -202,7 +204,6 @@ public class PacketCopier implements PacketInterceptor, ComponentEventListener {
     }
 
     private static class Subscription {
-        private Component component;
         private boolean presenceEnabled;
         private boolean messageEnabled;
         private boolean iqEnabled;
@@ -213,16 +214,14 @@ public class PacketCopier implements PacketInterceptor, ComponentEventListener {
          * 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,
+        public Subscription(boolean iqEnabled, boolean messageEnabled,
                 boolean presenceEnabled, boolean incoming, boolean processed) {
-            this.component = component;
             this.incoming = incoming;
             this.iqEnabled = iqEnabled;
             this.messageEnabled = messageEnabled;
@@ -230,15 +229,6 @@ public class PacketCopier implements PacketInterceptor, ComponentEventListener {
             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.

src/java/org/jivesoftware/openfire/pubsub/PubSubModule.java

@@ -193,7 +193,7 @@ public class PubSubModule extends BasicModule implements ServerItemsProvider, Di
 
     public boolean isServiceAdmin(JID user) {
         return sysadmins.contains(user.toBareJID()) || allowedToCreate.contains(user.toBareJID()) ||
-                InternalComponentManager.getInstance().getComponent(user) != null;
+                InternalComponentManager.getInstance().hasComponent(user);
     }
 
     public boolean isInstantNodeSupported() {

src/java/org/jivesoftware/openfire/session/LocalComponentSession.java

@@ -122,7 +122,7 @@ public class LocalComponentSession extends LocalSession implements ComponentSess
             return null;
         }
         // Check that the requested subdomain is not already in use
-        if (InternalComponentManager.getInstance().getComponent(subdomain) != null) {
+        if (InternalComponentManager.getInstance().hasComponent(new JID(subdomain + "." + serverName))) {
             Log.debug("[ExComp] Another component is already using domain: " + domain);
             // Domain already occupied so return a conflict error and close the connection
             // Include the conflict error in the response

src/java/org/jivesoftware/openfire/spi/PresenceManagerImpl.java

@@ -15,13 +15,7 @@ import org.dom4j.Document;
 import org.dom4j.DocumentException;
 import org.dom4j.DocumentHelper;
 import org.jivesoftware.database.DbConnectionManager;
-import org.jivesoftware.util.*;
-import org.jivesoftware.util.cache.Cache;
-import org.jivesoftware.util.cache.CacheFactory;
-import org.jivesoftware.openfire.PacketDeliverer;
-import org.jivesoftware.openfire.PresenceManager;
-import org.jivesoftware.openfire.SessionManager;
-import org.jivesoftware.openfire.XMPPServer;
+import org.jivesoftware.openfire.*;
 import org.jivesoftware.openfire.auth.UnauthorizedException;
 import org.jivesoftware.openfire.component.InternalComponentManager;
 import org.jivesoftware.openfire.container.BasicModule;
@@ -35,11 +29,16 @@ import org.jivesoftware.openfire.session.ClientSession;
 import org.jivesoftware.openfire.user.User;
 import org.jivesoftware.openfire.user.UserManager;
 import org.jivesoftware.openfire.user.UserNotFoundException;
-import org.xmpp.component.Component;
+import org.jivesoftware.util.LocaleUtils;
+import org.jivesoftware.util.Log;
+import org.jivesoftware.util.StringUtils;
+import org.jivesoftware.util.cache.Cache;
+import org.jivesoftware.util.cache.CacheFactory;
 import org.xmpp.packet.JID;
 import org.xmpp.packet.PacketError;
 import org.xmpp.packet.Presence;
 
+import java.sql.Connection;
 import java.sql.*;
 import java.util.ArrayList;
 import java.util.Collection;
@@ -63,6 +62,7 @@ public class PresenceManagerImpl extends BasicModule implements PresenceManager
     private static final String NULL_STRING = "NULL";
     private static final long NULL_LONG = -1L;
 
+    private RoutingTable routingTable;
     private SessionManager sessionManager;
     private UserManager userManager;
     private RosterManager rosterManager;
@@ -390,15 +390,14 @@ public class PresenceManagerImpl extends BasicModule implements PresenceManager
                 }
             }
             else {
-                Component component = getComponent(probee);
-                if (component != null) {
+                if (routingTable.hasComponentRoute(probee)) {
                     // If the probee belongs to a component then ask the component to process the
                     // probe presence
                     Presence presence = new Presence();
                     presence.setType(Presence.Type.probe);
                     presence.setFrom(prober);
                     presence.setTo(probee);
-                    component.processPacket(presence);
+                    routingTable.routePacket(probee, presence);
                 }
                 else {
                     // Check if the probee may be hosted by this server
@@ -477,6 +476,7 @@ public class PresenceManagerImpl extends BasicModule implements PresenceManager
         userManager = server.getUserManager();
         presenceUpdateHandler = server.getPresenceUpdateHandler();
         rosterManager = server.getRosterManager();
+        routingTable = server.getRoutingTable();
     }
 
     public void start() throws IllegalStateException {
@@ -492,15 +492,6 @@ public class PresenceManagerImpl extends BasicModule implements PresenceManager
         lastActivityCache.clear();
     }
 
-    private Component getComponent(JID probee) {
-        // Check for registered components
-        Component component = componentManager.getComponent(probee);
-        if (component != null) {
-            return component;
-        }
-        return null;
-    }
-
     /**
      * Loads offline presence data for the user into cache.
      *