basetransport.java

来自「基于Jabber协议的即时消息服务器」· Java 代码 · 共 1,590 行 · 第 1/5 页

/**
 * $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.gateway;

import org.dom4j.DocumentHelper;
import org.dom4j.Element;
import org.dom4j.QName;
import org.jivesoftware.util.Log;
import org.jivesoftware.util.NotFoundException;
import org.jivesoftware.wildfire.ClientSession;
import org.jivesoftware.wildfire.SessionManager;
import org.jivesoftware.wildfire.XMPPServer;
import org.jivesoftware.wildfire.container.PluginManager;
import org.jivesoftware.wildfire.roster.Roster;
import org.jivesoftware.wildfire.roster.*;
import org.jivesoftware.wildfire.user.PresenceEventListener;
import org.jivesoftware.wildfire.user.UserAlreadyExistsException;
import org.jivesoftware.wildfire.user.UserNotFoundException;
import org.jivesoftware.wildfire.user.PresenceEventDispatcher;
import org.xmpp.component.Component;
import org.xmpp.component.ComponentManager;
import org.xmpp.forms.DataForm;
import org.xmpp.forms.FormField;
import org.xmpp.packet.*;
import org.xmpp.packet.PacketError.Condition;

import java.util.*;

/**
 * Base class of all transport implementations.
 *
 * Handles all transport non-specific tasks and provides the glue that holds
 * together server interactions and the legacy service.  Does the bulk of
 * the XMPP related work.  Also note that this represents the transport
 * itself, not an individual session with the transport.
 *
 * @author Daniel Henninger
 */
public abstract class BaseTransport implements Component, RosterEventListener, PresenceEventListener {

    /**
     * Create a new BaseTransport instance.
     */
    public BaseTransport() {
        // We've got nothing to do here.
    }

    /**
     * Set up the transport instance.
     *
     * @param type Type of the transport.
     * @param description Description of the transport (for Disco).
     */
    public void setup(TransportType type, String description) {
        this.description = description;
        this.transportType = type;
    }

    /**
     * Handles initialization of the transport.
     */
    public void initialize(JID jid, ComponentManager componentManager) {
        this.jid = jid;
        this.componentManager = componentManager;
        sessionManager.startThreadManager(jid);
    }

    /**
     * Manages all active sessions.
     * @see org.jivesoftware.wildfire.gateway.TransportSessionManager
     */
    public final TransportSessionManager sessionManager = new TransportSessionManager(this);

    /**
     * Manages registration information.
     * @see org.jivesoftware.wildfire.gateway.RegistrationManager
     */
    public final RegistrationManager registrationManager = new RegistrationManager();

    /**
     * JID of the transport in question.
     */
    public JID jid = null;

    /**
     * Description of the transport in question.
     */
    public String description = null;

    /**
     * Component Manager associated with transport.
     */
    public ComponentManager componentManager = null;

    /**
     * Manager component for user rosters.
     */
    public final RosterManager rosterManager = new RosterManager();

    /**
     * Type of the transport in question.
     * @see org.jivesoftware.wildfire.gateway.TransportType
     */
    public TransportType transportType = null;

    private final String DISCO_INFO = "http://jabber.org/protocol/disco#info";
    private final String DISCO_ITEMS = "http://jabber.org/protocol/disco#items";
    private final String IQ_GATEWAY = "jabber:iq:gateway";
    private final String IQ_REGISTER = "jabber:iq:register";
    private final String IQ_VERSION = "jabber:iq:version";

    /**
     * Handles all incoming XMPP stanzas, passing them to individual
     * packet type handlers.
     *
     * @param packet The packet to be processed.
     */
    public void processPacket(Packet packet) {
        try {
            List<Packet> reply = new ArrayList<Packet>();
            if (packet instanceof IQ) {
                reply.addAll(processPacket((IQ)packet));
            }
            else if (packet instanceof Presence) {
                reply.addAll(processPacket((Presence)packet));
            }
            else if (packet instanceof Message) {
                reply.addAll(processPacket((Message)packet));
            }
            else {
                Log.info("Received an unhandled packet: " + packet.toString());
            }

            if (reply.size() > 0) {
                for (Packet p : reply) {
                    this.sendPacket(p);
                }
            }
        }
        catch (Exception e) {
            Log.error("Error occured while processing packet:", e);
        }
    }

    /**
     * Handles all incoming message stanzas.
     *
     * @param packet The message packet to be processed.
     * @return list of packets that will be sent back to the message sender.
     */
    private List<Packet> processPacket(Message packet) {
        Log.debug("Received message packet: "+packet.toXML());
        List<Packet> reply = new ArrayList<Packet>();
        JID from = packet.getFrom();
        JID to = packet.getTo();

        try {
            TransportSession session = sessionManager.getSession(from);
            if (!session.isLoggedIn()) {
                Message m = new Message();
                m.setError(Condition.service_unavailable);
                m.setTo(from);
                m.setFrom(getJID());
                m.setBody("You are not currently logged into the transport.");
                reply.add(m);
            }
            else if (to.getNode() == null) {
                // Message to gateway itself.  Throw away for now.
                if (packet.getBody() != null) {
                    session.sendServerMessage(packet.getBody());
                }
            }
            else {
                if (packet.getBody() != null) {
                    session.sendMessage(to, packet.getBody());
                }
            }
        }
        catch (NotFoundException e) {
            Log.debug("Unable to find session.");
            Message m = new Message();
            m.setError(Condition.service_unavailable);
            m.setTo(from);
            m.setFrom(getJID());
            m.setBody("You are not currently logged into the transport.");
            reply.add(m);
        }

        return reply;
    }

    /**
     * Handles all incoming presence stanzas.
     *
     * @param packet The presence packet to be processed.
     * @return list of packets that will be sent back to the presence requester.
     */
    private List<Packet> processPacket(Presence packet) {
        Log.debug("Received presence packet: "+packet.toXML());
        List<Packet> reply = new ArrayList<Packet>();
        JID from = packet.getFrom();
        JID to = packet.getTo();

        if (packet.getType() == Presence.Type.error) {
            // We don't want to do anything with this.  Ignore it.
            return reply;
        }

        try {
            if (to.getNode() == null) {
                Collection<Registration> registrations = registrationManager.getRegistrations(from, this.transportType);
                if (registrations.isEmpty()) {
                    // User is not registered with us.
                    Log.debug("Unable to find registration.");
                    return reply;
                }

                // This packet is to the transport itself.
                if (packet.getType() == Presence.Type.probe) {
                    // Client is asking for presence status.
                    TransportSession session;
                    try {
                        session = sessionManager.getSession(from);
                        if (session.isLoggedIn()) {
                            Presence p = new Presence();
                            p.setTo(from);
                            p.setFrom(to);
                            this.sendPacket(p);
                        }
                    }
                    catch (NotFoundException e) {
                        Log.debug("Ignoring probe presence for inactive session.");
                    }
                }
                else {
                    Log.debug("Ignoring this packet:" + packet.toString());
                    // Anything else we will ignore for now.
                }
            }
            else {
                // This packet is to a user at the transport.
                try {
                    TransportSession session = sessionManager.getSession(from);

                    if (packet.getType() == Presence.Type.probe) {
                        // Presence probe, lets try to tell them.
                        session.retrieveContactStatus(to);
                    }
                    else {
                        // Anything else we will ignore for now.
                    }
                }
                catch (NotFoundException e) {
                    // Well we just don't care then.
                }
            }
        }
        catch (Exception e) {
            Log.error("Exception while processing packet: ", e);
        }

        return reply;
    }

    /**
     * Handles all incoming iq stanzas.
     *
     * @param packet The iq packet to be processed.
     * @return list of packets that will be sent back to the IQ requester.
     */
    private List<Packet> processPacket(IQ packet) {
        Log.debug("Received iq packet: "+packet.toXML());                
        List<Packet> reply = new ArrayList<Packet>();

        if (packet.getType() == IQ.Type.error) {
            // Lets not start a loop.  Ignore.
            return reply;
        }

        String xmlns = null;
        Element child = (packet).getChildElement();
        if (child != null) {
            xmlns = child.getNamespaceURI();
        }

        if (xmlns == null) {
            // No namespace defined.
            Log.debug("No XMLNS:" + packet.toString());
            IQ error = IQ.createResultIQ(packet);
            error.setError(Condition.bad_request);
            reply.add(error);
            return reply;
        }

        if (xmlns.equals(DISCO_INFO)) {
            reply.addAll(handleDiscoInfo(packet));
        }
        else if (xmlns.equals(DISCO_ITEMS)) {
            reply.addAll(handleDiscoItems(packet));
        }
        else if (xmlns.equals(IQ_GATEWAY)) {
            reply.addAll(handleIQGateway(packet));
        }
        else if (xmlns.equals(IQ_REGISTER)) {
            reply.addAll(handleIQRegister(packet));
        }
        else if (xmlns.equals(IQ_VERSION)) {
            reply.addAll(handleIQVersion(packet));
        }
        else {