multicastrouter.java

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

/**
 * $RCSfile: $
 * $Revision: 2705 $
 * $Date: 2005-08-22 19:00:05 -0300 (Mon, 22 Aug 2005) $
 *
 * Copyright (C) 2005 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;

import org.dom4j.Element;
import org.jivesoftware.util.Cache;
import org.jivesoftware.util.CacheManager;
import org.jivesoftware.util.JiveConstants;
import org.jivesoftware.wildfire.container.BasicModule;
import org.jivesoftware.wildfire.disco.ServerFeaturesProvider;
import org.xmpp.packet.IQ;
import org.xmpp.packet.JID;
import org.xmpp.packet.Packet;

import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CopyOnWriteArrayList;

/**
 * Router of packets with multiple recipients. Clients may send a single packet with multiple
 * recipients and the server will broadcast the packet to the target receipients. If recipients
 * belong to remote servers, then this server will discover if remote target servers support
 * multicast service. If a remote server supports the multicast service, a single packet will be
 * sent to the remote server. If a remote server doesn't the support multicast
 * processing, the local server sends a copy of the original stanza to each address.<p>
 *
 * The current implementation will only search up to the first level of nodes of remote servers
 * when trying to find out if remote servers have support for multicast service. It is assumed
 * that it is highly unlikely for servers to have a node in the second or third depth level
 * providing the multicast service. Servers should normally provide this service themselves or
 * at least as a first level node.
 *
 * This is an implementation of <a href=http://www.jabber.org/jeps/jep-0033.html>
 * JEP-0033: Extended Stanza Addressing</a>
 *
 * @author Matt Tucker
 */
public class MulticastRouter extends BasicModule implements ServerFeaturesProvider, IQResultListener {

    private static final String NAMESPACE = "http://jabber.org/protocol/address";

    private XMPPServer server;
    /**
     * Router used for delivering packets with multiple recipients.
     */
    private PacketRouter packetRouter;
    /**
     * Router used for discovering if remote servers support multicast service.
     */
    private IQRouter iqRouter;
    /**
     * Cache for a day discovered information of remote servers. The local server will try
     * to discover if remote servers support multicast service.
     */
    private Cache cache;
    /**
     * Packets that include recipients that belong to remote servers are not processed by
     * the main thread since extra work is required. This variable holds the list of packets
     * pending to be sent to remote servers. Note: key=domain, value=collection of packet
     * pending to be sent.
     */
    private Map<String, Collection<Packet>> remotePackets =
            new HashMap<String, Collection<Packet>>();
    /**
     * Keeps the list of nodes discovered in remote servers. This information is used
     * when discovering whether remote servers support multicast service or not.
     * Note: key=domain, value=list of nodes
     */
    private Map<String, Collection<String>> nodes = new ConcurrentHashMap<String, Collection<String>>();
    /**
     * Keeps an association of node and server where the node was discovered. This information
     * is used when discovering whether remote servers support multicast service or not.
     * Note: key=node, value=domain of remote server
     */
    private Map<String, String> roots = new ConcurrentHashMap<String, String>();

    public MulticastRouter() {
        super("Multicast Packet Router");

        String cacheName = "Multicast Service";
        CacheManager.initializeCache(cacheName, "multicast", 128 * 1024, JiveConstants.DAY);
        cache = CacheManager.getCache(cacheName);
    }

    public void route(Packet packet) {
        Set<String> remoteServers = new HashSet<String>();
        List<String> targets = new ArrayList<String>();
        Packet localBroadcast = packet.createCopy();
        Element addresses = getAddresses(localBroadcast);
        String localDomain = "@" + server.getServerInfo().getName();
        // Build the <addresses> element to be included for local users and identify
        // remote domains that should receive the packet too
        for (Iterator it=addresses.elementIterator("address");it.hasNext();) {
            Element address = (Element) it.next();
            // Skip addresses of type noreply since they don't have any address
            if (Type.noreply.toString().equals(address.attributeValue("type"))) {
                continue;
            }
            String jid = address.attributeValue("jid");
            // Only send to local users and if packet has not already been delivered
            if (jid.contains(localDomain) && address.attributeValue("delivered") == null) {
                targets.add(jid);
            }
            else if (!jid.contains(localDomain)) {
                remoteServers.add(new JID(jid).getDomain());
            }
            // Set as delivered
            address.addAttribute("delivered", "true");
            // Remove bcc addresses
            if (Type.bcc.toString().equals(address.attributeValue("type"))) {
                it.remove();
            }
        }
        // Send the packet to local target users
        for (String jid : targets) {
            localBroadcast.setTo(jid);
            packetRouter.route(localBroadcast);
        }

        // Keep a registry of packets that should be sent to remote domains.
        for (String domain : remoteServers) {
            boolean shouldDiscover = false;
            synchronized (domain.intern()) {
                Collection<Packet> packets = remotePackets.get(domain);
                if (packets == null) {
                    packets = new ArrayList<Packet>();
                    remotePackets.put(domain, packets);
                    shouldDiscover = true;
                }
                // Add that this packet should be sent to the requested remote server
                packets.add(packet);
            }
            if (shouldDiscover) {
                // First time a packet is sent to this remote server so start the extra work
                // of discovering if remote server supports multicast service and actually send
                // the packet to the remote server
                sendToRemoteEntity(domain);
            }
        }
        // TODO Add thread that checks every 5 minutes if packets to remote servers were not
        // TODO sent because no disco response was received. So assume that remote server does
        // TODO not support JEP-33 and send pending packets
    }

    /**
     * Returns the Element that contains the multiple recipients.
     *
     * @param packet packet containing the multiple recipients.
     * @return the Element that contains the multiple recipients.
     */
    private Element getAddresses(Packet packet) {
        if (packet instanceof IQ) {
            return ((IQ) packet).getChildElement().element("addresses");
        }
        else {
            return packet.getElement().element("addresses");
        }
    }

    /**
     * Sends pending packets of the requested domain but first try to discover if remote server
     * supports multicast service. If we already have cached information about the requested
     * domain then just deliver the packet.
     *
     * @param domain the domain that has pending packets to be sent.
     */
    private void sendToRemoteEntity(String domain) {
        // Check if there is cached information about the requested domain
        String multicastService = (String) cache.get(domain);
        if (multicastService != null) {
            sendToRemoteServer(domain, multicastService);
        }
        else {
            // No cached information was found so discover if remote server
            // supports JEP-33 (Extended Stanza Addressing). The reply to the disco
            // request is going to be process in #receivedAnswer(IQ packet)
            IQ iq = new IQ(IQ.Type.get);
            iq.setFrom(server.getServerInfo().getName());
            iq.setTo(domain);
            iq.setChildElement("query", "http://jabber.org/protocol/disco#info");
            // Indicate that we are searching for info of the specified domain
            nodes.put(domain, new CopyOnWriteArrayList<String>());
            // Send the disco#info request to the remote server or component. The reply will be
            // processed by the IQResultListener (interface that this class implements)
            iqRouter.addIQResultListener(iq.getID(), this);
            iqRouter.route(iq);
        }
    }

    /**
     * Actually sends pending packets of the specified domain using the discovered multicast
     * service address. If remote server supports multicast service then a copy of the
     * orignal will be sent to the remote server. However, if no multicast service was found
     * then the local server sends a copy of the original stanza to each address.
     *
     * @param domain domain of the remote server with pending packets.
     * @param multicastService address of the discovered multicast service.
     */
    private void sendToRemoteServer(String domain, String multicastService) {
        Collection<Packet> packets = null;
        // Get the packets to send to the remote entity
        synchronized (domain.intern()) {
            packets = remotePackets.remove(domain);
        }

        if (multicastService != null && multicastService.trim().length() > 0) {