Initial version. JM-666

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

src/java/org/jivesoftware/wildfire/multiplex/ClientSessionConnection.java

+/**
+ * $RCSfile: $
+ * $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.multiplex;
+
+import org.dom4j.Element;
+import org.jivesoftware.util.StringUtils;
+import org.jivesoftware.wildfire.XMPPServer;
+import org.jivesoftware.wildfire.net.VirtualConnection;
+import org.xmpp.packet.IQ;
+import org.xmpp.packet.Packet;
+
+import java.net.InetAddress;
+import java.net.UnknownHostException;
+
+/**
+ * Represents a connection of a Client Session that was established to a Connection Manager.
+ * Connection Managers have their own physical connections to the server that are multiplexed
+ * among connected clients. Each created {@link org.jivesoftware.wildfire.ClientSession} will
+ * use an instance of this class as its connection.
+ *
+ * @author Gaston Dombiak
+ */
+public class ClientSessionConnection extends VirtualConnection {
+
+    private String connectionManagerName;
+    private String serverName;
+    private ConnectionMultiplexerManager multiplexerManager;
+
+    public ClientSessionConnection(String connectionManagerName) {
+        this.connectionManagerName = connectionManagerName;
+        multiplexerManager = ConnectionMultiplexerManager.getInstance();
+        serverName = XMPPServer.getInstance().getServerInfo().getName();
+    }
+
+    /**
+     * Delivers the packet to the Connection Manager that in turn will forward it to the
+     * target user. Connection Managers may have one or many connections to the server so
+     * just get any connection to the Connection Manager (uses a random) and use it.<p>
+     *
+     * If the packet to send does not have a TO attribute then wrap the packet with a
+     * special IQ packet. The wrapper IQ packet will be sent to the Connection Manager
+     * and the stream ID of this Client Session will be used for identifying that the wrapped
+     * packet must be sent to the connected user. Since some packets can be exchanged before
+     * the user has a binded JID we need to use the stream ID as the unique identifier.
+     *
+     * @param packet the packet to send to the user.
+     */
+    public void deliver(Packet packet) {
+        ConnectionMultiplexerSession multiplexerSession =
+                multiplexerManager.getMultiplexerSession(connectionManagerName);
+        if (multiplexerSession != null) {
+            // If TO is null then wrap packet so that the connection manager can
+            // figure out the target session
+            if (packet.getTo() == null) {
+                IQ wrapper = new IQ(IQ.Type.set);
+                wrapper.setFrom(serverName);
+                wrapper.setTo(connectionManagerName);
+                Element child = wrapper.setChildElement("session",
+                        "http://jabber.org/protocol/connectionmanager");
+                child.addAttribute("id", session.getStreamID().getID());
+                Element send = child.addElement("send");
+                send.add(packet.getElement().createCopy());
+                // Deliver wrapper
+                multiplexerSession.deliver(wrapper);
+            }
+            else {
+                // Deliver original packet
+                multiplexerSession.deliver(packet);
+            }
+            session.incrementServerPacketCount();
+        }
+    }
+
+    /**
+     * Delivers the stanza to the Connection Manager that in turn will forward it to the
+     * target user. Connection Managers may have one or many connections to the server so
+     * just get any connection to the Connection Manager (uses a random) and use it.<p>
+     *
+     * The stanza to send wrapped with a special IQ packet. The wrapper IQ packet will be
+     * sent to the Connection Manager and the stream ID of this Client Session will be used
+     * for identifying that the wrapped stanza must be sent to the connected user.
+     *
+     * @param text the stanza to send to the user.
+     */
+    public void deliverRawText(String text) {
+        ConnectionMultiplexerSession multiplexerSession =
+                multiplexerManager.getMultiplexerSession(connectionManagerName);
+        if (multiplexerSession != null) {
+            // Wrap packet so that the connection manager can figure out the target session
+            StringBuilder sb = new StringBuilder(200 + text.length());
+            sb.append("<iq type=\"set\" from=\"").append(serverName);
+            sb.append("\" to=\"").append(connectionManagerName);
+            sb.append("\" id=\"").append(StringUtils.randomString(10));
+            sb.append("\"><session xmlns=\"http://jabber.org/protocol/connectionmanager\" id=\"");
+            sb.append(session.getStreamID().getID()).append("\"><send>");
+            sb.append(text);
+            sb.append("</send></session></iq>");
+            // Deliver the wrapped stanza
+            multiplexerSession.getConnection().deliverRawText(sb.toString());
+        }
+    }
+
+    public InetAddress getInetAddress() throws UnknownHostException {
+        //TODO Future version may return actual IP client address. We would need to pass this info
+        // Return IP address of the connection manager that the client used to log in
+        ConnectionMultiplexerSession multiplexerSession =
+                multiplexerManager.getMultiplexerSession(connectionManagerName);
+        if (multiplexerSession != null) {
+            return multiplexerSession.getConnection().getInetAddress();
+        }
+        return null;
+    }
+
+    /**
+     * If the Connection Manager or the Client requested to close the connection then just do
+     * nothing. But if the server originated the request to close the connection then we need
+     * to send to the Connection Manager a packet letting him know that the Client Session needs
+     * to be terminated.
+     */
+    public void closeVirtualConnection() {
+        // Figure out who requested the connection to be closed
+        String streamID = session.getStreamID().getID();
+        if (multiplexerManager.getClientSession(connectionManagerName, streamID) == null) {
+            // Client or Connection manager requested to close the session
+            // Do nothing since it has already been removed and closed
+        }
+        else {
+            // Server requested to close the client session so let the connection manager
+            // know that he has to finish the client session
+            IQ closeRequest = new IQ(IQ.Type.set);
+            closeRequest.setFrom(serverName);
+            closeRequest.setTo(connectionManagerName);
+            Element child = closeRequest.setChildElement("session",
+                    "http://jabber.org/protocol/connectionmanager");
+            child.addAttribute("id", streamID);
+            child.addElement("close");
+            deliver(closeRequest);
+        }
+    }
+}

src/java/org/jivesoftware/wildfire/multiplex/ConnectionMultiplexerManager.java

+/**
+ * $RCSfile: $
+ * $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.multiplex;
+
+import org.jivesoftware.util.Log;
+import org.jivesoftware.wildfire.*;
+import org.jivesoftware.wildfire.auth.UnauthorizedException;
+import org.jivesoftware.wildfire.event.SessionEventDispatcher;
+import org.jivesoftware.wildfire.event.SessionEventListener;
+
+import java.util.List;
+import java.util.Map;
+import java.util.Random;
+import java.util.concurrent.ConcurrentHashMap;
+
+/**
+ * A ConnectionMultiplexerManager is responsible for keeping track of the connected
+ * Connection Managers and the sessions that were established with the Connection
+ * Managers. Moreover, a ConnectionMultiplexerManager is able to create, get and close
+ * client sessions based on Connection requests.
+ *
+ * @author Gaston Dombiak
+ */
+public class ConnectionMultiplexerManager implements SessionEventListener {
+
+    private static final ConnectionMultiplexerManager instance = new ConnectionMultiplexerManager();
+
+    /**
+     * Pseudo-random number generator object for use with getMultiplexerSession(String).
+     */
+    private static Random randGen = new Random();
+
+    static {
+        // Add the unique instance of this class as a session listener. We need to react
+        // when sessions are closed so we can clean up the registry of client sessions.
+        SessionEventDispatcher.addListener(instance);
+    }
+
+    /**
+     * Map that keeps track of connection managers and hosted connections.
+     * Key: stream ID; Value: Domain of connection manager hosting connection
+     */
+    private Map<String, String> streamIDs = new ConcurrentHashMap<String, String>();
+    /**
+     * Map that keeps track of connection managers and hosted sessions.
+     * Key: Domain of connection manager; Value: Map with Key: stream ID; Value: Client session
+     */
+    private Map<String, Map<String, ClientSession>> sessionsByManager =
+            new ConcurrentHashMap<String, Map<String, ClientSession>>();
+
+    private SessionManager sessionManager;
+
+    /**
+     * Returns the unique instance of this class.
+     *
+     * @return the unique instance of this class.
+     */
+    public static ConnectionMultiplexerManager getInstance() {
+        return instance;
+    }
+
+    private ConnectionMultiplexerManager() {
+        sessionManager = XMPPServer.getInstance().getSessionManager();
+    }
+
+    /**
+     * Creates a new client session that was established to the specified connection manager.
+     * The new session will not be findable through its stream ID.
+     *
+     * @param connectionManagerDomain the connection manager that is handling the connection
+     *        of the session.
+     * @param streamID the stream ID created by the connection manager for the new session.
+     */
+    public void createClientSession(String connectionManagerDomain, String streamID) {
+        try {
+            Connection connection = new ClientSessionConnection(connectionManagerDomain);
+            ClientSession session = SessionManager.getInstance()
+                    .createClientSession(connection, new BasicStreamID(streamID));
+            // Register that this streamID belongs to the specified connection manager
+            streamIDs.put(streamID, connectionManagerDomain);
+            // Register which sessions are being hosted by the speicifed connection manager
+            Map<String, ClientSession> sessions = sessionsByManager.get(connectionManagerDomain);
+            if (sessions == null) {
+                synchronized (connectionManagerDomain.intern()) {
+                    sessions = sessionsByManager.get(connectionManagerDomain);
+                    if (sessions == null) {
+                        sessions = new ConcurrentHashMap<String, ClientSession>();
+                        sessionsByManager.put(connectionManagerDomain, sessions);
+                    }
+                }
+            }
+            sessions.put(streamID, session);
+        }
+        catch (UnauthorizedException e) {
+            Log.error("Error creating virtual client session", e);
+        }
+    }
+
+    /**
+     * Closes an existing client session that was established through a connection manager.
+     *
+     * @param connectionManagerDomain the connection manager that is handling the connection
+     *        of the session.
+     * @param streamID the stream ID created by the connection manager for the session.
+     */
+    public void closeClientSession(String connectionManagerDomain, String streamID) {
+        Map<String, ClientSession> sessions = sessionsByManager.get(connectionManagerDomain);
+        if (sessions != null) {
+            Session session = sessions.remove(streamID);
+            if (session != null) {
+                // Close the session
+                session.getConnection().close();
+            }
+        }
+    }
+
+    /**
+     * Close client sessions that were established to the specified connection manager. This
+     * action is usually required when the connection manager was stopped or suddenly went
+     * down.
+     *
+     * @param connectionManagerDomain the connection manager that is no longer available.
+     */
+    public void closeClientSessions(String connectionManagerDomain) {
+        // Remove the connection manager and the hosted sessions
+        Map<String, ClientSession> sessions = sessionsByManager.remove(connectionManagerDomain);
+        if (sessions != null) {
+            for (String streamID : sessions.keySet()) {
+                // Remove inverse track of connection manager hosting streamIDs
+                streamIDs.remove(streamID);
+                // Close the session
+                sessions.get(streamID).getConnection().close();
+            }
+        }
+    }
+
+    /**
+     * Returns the ClientSession with the specified stream ID that is being hosted by the
+     * specified connection manager.
+     *
+     * @param connectionManagerDomain the connection manager that is handling the connection
+     *        of the session.
+     * @param streamID the stream ID created by the connection manager for the session.
+     * @return the ClientSession with the specified stream ID.
+     */
+    public ClientSession getClientSession(String connectionManagerDomain, String streamID) {
+        Map<String, ClientSession> sessions = sessionsByManager.get(connectionManagerDomain);
+        if (sessions != null) {
+            return sessions.get(streamID);
+        }
+        return null;
+    }
+
+    /**
+     * Returns a {@link ConnectionMultiplexerSession} for the specified connection manager
+     * domain or <tt>null</tt> if none was found. In case the connection manager has many
+     * connections established with the server then one of them will be selected randomly.
+     *
+     * @param connectionManagerDomain the domain of the connection manager to get a session.
+     * @return a session to the specified connection manager domain or null if none was found.
+     */
+    public ConnectionMultiplexerSession getMultiplexerSession(String connectionManagerDomain) {
+        List<ConnectionMultiplexerSession> sessions =
+                sessionManager.getConnectionMultiplexerSessions(connectionManagerDomain);
+        if (sessions.isEmpty()) {
+            return null;
+        }
+        else if (sessions.size() == 1) {
+            return sessions.get(0);
+        }
+        else {
+            // Pick a random session so we can distribute traffic evenly
+            return sessions.get(randGen.nextInt(sessions.size()));
+        }
+    }
+
+    public void anonymousSessionCreated(Session session) {
+        // Do nothing.
+    }
+
+    public void anonymousSessionDestroyed(Session session) {
+        removeSession(session);
+    }
+
+    public void sessionCreated(Session session) {
+        // Do nothing.
+    }
+
+    public void sessionDestroyed(Session session) {
+        removeSession(session);
+    }
+
+    private void removeSession(Session session) {
+        // Remove trace indicating that a connection manager is hosting a connection
+        String streamID = session.getStreamID().getID();
+        String connectionManagerDomain = streamIDs.remove(streamID);
+        // Remove trace indicating that a connection manager is hosting a session
+        if (connectionManagerDomain != null) {
+            Map<String, ClientSession> sessions = sessionsByManager.get(connectionManagerDomain);
+            if (sessions != null) {
+                sessions.remove(streamID);
+            }
+        }
+    }
+
+    /**
+     * Simple implementation of the StreamID interface to hold the stream ID assigned by
+     * the Connection Manager to the Session.
+     */
+    private class BasicStreamID implements StreamID {
+        String id;
+
+        public BasicStreamID(String id) {
+            this.id = id;
+        }
+
+        public String getID() {
+            return id;
+        }
+
+        public String toString() {
+            return id;
+        }
+
+        public int hashCode() {
+            return id.hashCode();
+        }
+    }
+}

src/java/org/jivesoftware/wildfire/multiplex/MultiplexerPacketDeliverer.java

+/**
+ * $RCSfile: $
+ * $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.multiplex;
+
+import org.dom4j.Element;
+import org.jivesoftware.util.LocaleUtils;
+import org.jivesoftware.util.Log;
+import org.jivesoftware.wildfire.OfflineMessageStrategy;
+import org.jivesoftware.wildfire.PacketDeliverer;
+import org.jivesoftware.wildfire.PacketException;
+import org.jivesoftware.wildfire.XMPPServer;
+import org.jivesoftware.wildfire.auth.UnauthorizedException;
+import org.xmpp.packet.IQ;
+import org.xmpp.packet.Message;
+import org.xmpp.packet.Packet;
+import org.xmpp.packet.Presence;
+
+/**
+ * Fallback method used by {@link org.jivesoftware.wildfire.net.SocketConnection} when
+ * connected to a connection manager. The fallback method will be used when a SocketConnection
+ * fails to send a {@link Packet} (probably because the socket was closed).<p>
+ *
+ * The first attempt will be to send the packet using another connection to the same connection
+ * manager (since managers may have a pool of connections to the server). And if that fails then
+ * instances of {@link Message} may be stored offline for later retrieval. Since packets may be
+ * wrapped by special IQ packets (read the Connection Manager JEP for more information) we need
+ * to unwrap the packet and store the wrapped packet offline.
+ *
+ * @author Gaston Dombiak
+ */
+public class MultiplexerPacketDeliverer implements PacketDeliverer {
+
+    private OfflineMessageStrategy messageStrategy;
+    private String connectionManagerDomain;
+    private ConnectionMultiplexerManager multiplexerManager;
+
+    public MultiplexerPacketDeliverer() {
+        this.messageStrategy = XMPPServer.getInstance().getOfflineMessageStrategy();
+        multiplexerManager = ConnectionMultiplexerManager.getInstance();
+    }
+
+    void setConnectionManagerDomain(String connectionManagerDomain) {
+        this.connectionManagerDomain = connectionManagerDomain;
+    }
+
+    public void deliver(Packet packet) throws UnauthorizedException, PacketException {
+        // Check if we can send the packet using another session
+        if (connectionManagerDomain == null) {
+            // Packet deliverer has not yet been configured so handle unprocessed packet
+            handleUnprocessedPacket(packet);
+        }
+        else {
+            // Try getting another session to the same connection manager 
+            ConnectionMultiplexerSession session =
+                    multiplexerManager.getMultiplexerSession(connectionManagerDomain);
+            if (session == null || session.getConnection().isClosed()) {
+                // No other session was found so handle unprocessed packet
+                handleUnprocessedPacket(packet);
+            }
+            else {
+                // Send the packet using this other session to the same connection manager
+                session.deliver(packet);
+            }
+        }
+    }
+
+    private void handleUnprocessedPacket(Packet packet) {
+        if (packet instanceof Message) {
+            messageStrategy.storeOffline((Message) packet);
+        }
+        else if (packet instanceof Presence) {
+            // presence packets are dropped silently
+            //dropPacket(packet);
+        }
+        else if (packet instanceof IQ) {
+            IQ iq = (IQ) packet;
+            // Check if we need to unwrap the packet
+            Element child = iq.getChildElement();
+            if (child != null && "session".equals(child.getName()) &&
+                    "http://jabber.org/protocol/connectionmanager"
+                            .equals(child.getNamespacePrefix())) {
+                Element send = child.element("send");
+                if (send != null) {
+                    // Unwrap packet
+                    Element wrappedElement = (Element) send.elements().get(0);
+                    if ("message".equals(wrappedElement.getName())) {
+                        handleUnprocessedPacket(new Message(wrappedElement));
+                    }
+                }
+            }
+            else {
+                // IQ packets are logged but dropped
+                Log.warn(LocaleUtils.getLocalizedString("admin.error.routing") + "\n" +
+                        packet.toString());
+            }
+        }
+    }
+}