connectionmultiplexersession.java

基于Jabber协议的即时消息服务器

/**
 * $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.dom4j.io.XMPPPacketReader;
import org.jivesoftware.util.JiveGlobals;
import org.jivesoftware.util.LocaleUtils;
import org.jivesoftware.util.Log;
import org.jivesoftware.wildfire.*;
import org.jivesoftware.wildfire.auth.AuthFactory;
import org.jivesoftware.wildfire.auth.UnauthorizedException;
import org.jivesoftware.wildfire.net.SASLAuthentication;
import org.jivesoftware.wildfire.net.SocketConnection;
import org.xmlpull.v1.XmlPullParser;
import org.xmlpull.v1.XmlPullParserException;
import org.xmpp.packet.IQ;
import org.xmpp.packet.JID;
import org.xmpp.packet.Packet;
import org.xmpp.packet.StreamError;

import java.io.IOException;
import java.io.Writer;
import java.util.Collection;

/**
 * Represents a session between the server and a connection manager.<p>
 *
 * Each Connection Manager has its own domain. Each connection from the same connection manager
 * uses a different resource. Unlike any other session, connection manager sessions are not
 * present in the routing table. This means that connection managers are not reachable entities.
 * In other words, entities cannot send packets to connection managers but clients being hosted
 * by them. The main reason behind this design decision is that connection managers are private
 * components of the server so they can only be contacted by the server. Connection Manager
 * sessions are present in {@link SessionManager} but not in {@link RoutingTable}. Use
 * {@link SessionManager#getConnectionMultiplexerSessions(String)} to get all sessions or
 * {@link ConnectionMultiplexerManager#getMultiplexerSession(String)}
 * to get a random session to a given connection manager.
 *
 * @author Gaston Dombiak
 */
public class ConnectionMultiplexerSession extends Session {

    private static Connection.TLSPolicy tlsPolicy;
    private static Connection.CompressionPolicy compressionPolicy;

    /**
     * Milliseconds a connection has to be idle to be closed. Default is 30 minutes. Sending
     * stanzas to the client is not considered as activity. We are only considering the connection
     * active when the client sends some data or hearbeats (i.e. whitespaces) to the server.
     * The reason for this is that sending data will fail if the connection is closed. And if
     * the thread is blocked while sending data (because the socket is closed) then the clean up
     * thread will close the socket anyway.
     */
    private static long idleTimeout;

    static {
        // Set the TLS policy stored as a system property
        String policyName = JiveGlobals.getProperty("xmpp.multiplex.tls.policy",
                Connection.TLSPolicy.disabled.toString());
        tlsPolicy = Connection.TLSPolicy.valueOf(policyName);

        // Set the Compression policy stored as a system property
        policyName = JiveGlobals.getProperty("xmpp.multiplex.compression.policy",
                Connection.CompressionPolicy.disabled.toString());
        compressionPolicy = Connection.CompressionPolicy.valueOf(policyName);

        // Set the default read idle timeout. If none was set then assume 5 minutes
        idleTimeout = JiveGlobals.getIntProperty("xmpp.multiplex.idle", 5 * 60 * 1000);
    }

    public static Session createSession(String serverName, XMPPPacketReader reader,
            SocketConnection connection) throws XmlPullParserException, IOException,
            UnauthorizedException {
        XmlPullParser xpp = reader.getXPPParser();
        String domain = xpp.getAttributeValue("", "to");

        Log.debug("[ConMng] Starting registration of new connection manager for domain: " + domain);

        Writer writer = connection.getWriter();
        // Default answer header in case of an error
        StringBuilder sb = new StringBuilder();
        sb.append("<?xml version='1.0' encoding='");
        sb.append(CHARSET);
        sb.append("'?>");
        sb.append("<stream:stream ");
        sb.append("xmlns:stream=\"http://etherx.jabber.org/streams\" ");
        sb.append("xmlns=\"jabber:connectionmanager\" from=\"");
        sb.append(domain);
        sb.append("\" version=\"1.0\">");

        // Check that a domain was provided in the stream header
        if (domain == null) {
            Log.debug("[ConMng] Domain not specified in stanza: " + xpp.getText());
            // Include the bad-format in the response
            StreamError error = new StreamError(StreamError.Condition.bad_format);
            sb.append(error.toXML());
            writer.write(sb.toString());
            writer.flush();
            // Close the underlying connection
            connection.close();
            return null;
        }

        // Get the requested domain
        JID address = new JID(domain);
        // Check that a secret key was configured in the server
        String secretKey = ConnectionMultiplexerManager.getDefaultSecret();
        if (secretKey == null) {
            Log.debug("[ConMng] A shared secret for connection manager was not found.");
            // Include the internal-server-error in the response
            StreamError error = new StreamError(StreamError.Condition.internal_server_error);
            sb.append(error.toXML());
            writer.write(sb.toString());
            writer.flush();
            // Close the underlying connection
            connection.close();
            return null;
        }
        // Check that the requested subdomain is not already in use
        if (SessionManager.getInstance().getConnectionMultiplexerSession(address) != null) {
            Log.debug("[ConMng] Another connection manager is already using domain: " + domain);
            // Domain already occupied so return a conflict error and close the connection
            // Include the conflict error in the response
            StreamError error = new StreamError(StreamError.Condition.conflict);
            sb.append(error.toXML());
            writer.write(sb.toString());
            writer.flush();
            // Close the underlying connection
            connection.close();
            return null;
        }

        // Indicate the TLS policy to use for this connection
        connection.setTlsPolicy(tlsPolicy);

        // Indicate the compression policy to use for this connection
        connection.setCompressionPolicy(compressionPolicy);

        // Set the max number of milliseconds the connection may not receive data from the
        // client before closing the connection
        connection.setIdleTimeout(idleTimeout);

        // Set the connection manager domain to use delivering a packet fails
        ((MultiplexerPacketDeliverer) connection.getPacketDeliverer())
                .setConnectionManagerDomain(address.getDomain());

        // Create a ConnectionMultiplexerSession for the new session originated
        // from the connection manager
        Session session =
                SessionManager.getInstance().createMultiplexerSession(connection, address);
        // Set the address of the new session
        session.setAddress(address);

        try {
            Log.debug("[ConMng] Send stream header with ID: " + session.getStreamID() +
                    " for connection manager with domain: " +
                    domain);
            // Build the start packet response
            sb = new StringBuilder();
            sb.append("<?xml version='1.0' encoding='");
            sb.append(CHARSET);
            sb.append("'?>");
            sb.append("<stream:stream ");
            sb.append("xmlns:stream=\"http://etherx.jabber.org/streams\" ");
            sb.append("xmlns=\"jabber:connectionmanager\" from=\"");
            sb.append(domain);
            sb.append("\" id=\"");
            sb.append(session.getStreamID().toString());
            sb.append("\" version=\"1.0\" >");
            writer.write(sb.toString());
            writer.flush();

            // Announce stream features.

            sb = new StringBuilder(490);
            sb.append("<stream:features>");
            if (tlsPolicy != Connection.TLSPolicy.disabled) {
                sb.append("<starttls xmlns=\"urn:ietf:params:xml:ns:xmpp-tls\">");
                if (tlsPolicy == Connection.TLSPolicy.required) {
                    sb.append("<required/>");
                }
                sb.append("</starttls>");
            }
            // Include Stream features
            String specificFeatures = session.getAvailableStreamFeatures();
            if (specificFeatures != null) {
                sb.append(specificFeatures);
            }
            sb.append("</stream:features>");

            writer.write(sb.toString());
            writer.flush();