Initial version. Optimization. JM-925

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

src/java/org/jivesoftware/wildfire/net/MultiplexerStanzaHandler.java

+/**
+ * $Revision: $
+ * $Date: $
+ *
+ * Copyright (C) 2007 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.net;
+
+import org.dom4j.Element;
+import org.jivesoftware.wildfire.Connection;
+import org.jivesoftware.wildfire.PacketRouter;
+import org.jivesoftware.wildfire.auth.UnauthorizedException;
+import org.jivesoftware.wildfire.multiplex.MultiplexerPacketHandler;
+import org.jivesoftware.wildfire.multiplex.Route;
+import org.jivesoftware.wildfire.session.ConnectionMultiplexerSession;
+import org.jivesoftware.wildfire.session.Session;
+import org.xmlpull.v1.XmlPullParser;
+import org.xmlpull.v1.XmlPullParserException;
+import org.xmpp.packet.IQ;
+import org.xmpp.packet.Message;
+import org.xmpp.packet.PacketError;
+import org.xmpp.packet.Presence;
+
+/**
+ * Handler of XML stanzas sent by Connection Managers.
+ *
+ * @author Gaston Dombiak
+ */
+public class MultiplexerStanzaHandler extends StanzaHandler {
+
+    /**
+     * Handler of IQ packets sent from the Connection Manager to the server.
+     */
+    private MultiplexerPacketHandler packetHandler;
+
+    public MultiplexerStanzaHandler(PacketRouter router, String serverName, Connection connection) {
+        super(router, serverName, connection);
+    }
+
+    protected void processIQ(final IQ packet) {
+        if (session.getStatus() != Session.STATUS_AUTHENTICATED) {
+            // Session is not authenticated so return error
+            IQ reply = new IQ();
+            reply.setChildElement(packet.getChildElement().createCopy());
+            reply.setID(packet.getID());
+            reply.setTo(packet.getFrom());
+            reply.setFrom(packet.getTo());
+            reply.setError(PacketError.Condition.not_authorized);
+            session.process(reply);
+            return;
+        }
+        // Process the packet
+        packetHandler.handle(packet);
+    }
+
+    protected void processMessage(final Message packet) throws UnauthorizedException {
+        throw new UnauthorizedException("Message packets are not supported. Original packets " +
+                "should be wrapped by route packets.");
+    }
+
+    protected void processPresence(final Presence packet) throws UnauthorizedException {
+        throw new UnauthorizedException("Message packets are not supported. Original packets " +
+                "should be wrapped by route packets.");
+    }
+
+    /**
+     * Process stanza sent by a client that is connected to a connection manager. The
+     * original stanza is wrapped in the route element. Only a single stanza must be
+     * wrapped in the route element.
+     *
+     * @param packet the route element.
+     */
+    private void processRoute(final Route packet) {
+        if (session.getStatus() != Session.STATUS_AUTHENTICATED) {
+            // Session is not authenticated so return error
+            Route reply = new Route(packet.getStreamID());
+            reply.setID(packet.getID());
+            reply.setTo(packet.getFrom());
+            reply.setFrom(packet.getTo());
+            reply.setError(PacketError.Condition.not_authorized);
+            session.process(reply);
+            return;
+        }
+        // Process the packet
+        packetHandler.route(packet);
+    }
+
+    boolean processUnknowPacket(Element doc) {
+        String tag = doc.getName();
+        if ("route".equals(tag)) {
+            // Process stanza wrapped by the route packet
+            processRoute(new Route(doc));
+            return true;
+        }
+        else if ("handshake".equals(tag)) {
+            if (!((ConnectionMultiplexerSession)session).authenticate(doc.getStringValue())) {
+                session.getConnection().close();
+            }
+            return true;
+        }
+        else if ("error".equals(tag) && "stream".equals(doc.getNamespacePrefix())) {
+            session.getConnection().close();
+            return true;
+        }
+        return false;
+    }
+
+    String getNamespace() {
+        return "jabber:connectionmanager";
+    }
+
+    boolean validateHost() {
+        return false;
+    }
+
+    boolean createSession(String namespace, String serverName, XmlPullParser xpp, Connection connection)
+            throws XmlPullParserException {
+        if (getNamespace().equals(namespace)) {
+            // The connected client is a connection manager so create a ConnectionMultiplexerSession
+            session = ConnectionMultiplexerSession.createSession(serverName, xpp, connection);
+            if (session != null) {
+                packetHandler = new MultiplexerPacketHandler(session.getAddress().getDomain());
+            }
+            return true;
+        }
+        return false;
+    }
+}

src/java/org/jivesoftware/wildfire/net/StanzaHandler.java

+/**
+ * $Revision: $
+ * $Date: $
+ *
+ * Copyright (C) 2007 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.net;
+
+import org.dom4j.DocumentException;
+import org.dom4j.Element;
+import org.dom4j.io.XMPPPacketReader;
+import org.jivesoftware.util.LocaleUtils;
+import org.jivesoftware.util.Log;
+import org.jivesoftware.util.StringUtils;
+import org.jivesoftware.wildfire.Connection;
+import org.jivesoftware.wildfire.PacketRouter;
+import org.jivesoftware.wildfire.auth.UnauthorizedException;
+import org.jivesoftware.wildfire.session.Session;
+import org.xmlpull.v1.XmlPullParser;
+import org.xmlpull.v1.XmlPullParserException;
+import org.xmlpull.v1.XmlPullParserFactory;
+import org.xmpp.packet.*;
+
+import java.io.IOException;
+import java.io.StringReader;
+
+/**
+ * A StanzaHandler is the main responsible for handling incoming stanzas. Some stanzas like startTLS
+ * are totally managed by this class. The rest of the stanzas are just forwarded to the router.
+ *
+ * @author Gaston Dombiak
+ */
+public abstract class StanzaHandler {
+    /**
+     * The utf-8 charset for decoding and encoding Jabber packet streams.
+     */
+    protected static String CHARSET = "UTF-8";
+    private static final String STREAM_START = "<stream:stream";
+    /**
+     * Reuse the same factory for all the connections.
+     */
+    private static XmlPullParserFactory factory = null;
+
+    private Connection connection;
+
+    // DANIELE: Indicate if a session is already created
+    private boolean sessionCreated = false;
+
+    // Flag that indicates that the client requested to be authenticated. Once the
+    // authentication process is over the value will return to false.
+    private boolean startedSASL = false;
+
+    // DANIELE: Indicate if a stream:stream is arrived to complete compression
+    private boolean waitingCompressionACK = false;
+
+    /**
+     * Session associated with the socket reader.
+     */
+    protected Session session;
+    /**
+     * Server name for which we are attending clients.
+     */
+    private String serverName;
+
+    /**
+     * Router used to route incoming packets to the correct channels.
+     */
+    private PacketRouter router;
+
+    private XMPPPacketReader reader;
+
+    static {
+        try {
+            factory = XmlPullParserFactory.newInstance(MXParser.class.getName(), null);
+            factory.setNamespaceAware(true);
+        }
+        catch (XmlPullParserException e) {
+            Log.error("Error creating a parser factory", e);
+        }
+    }
+
+    /**
+     * Creates a dedicated reader for a socket.
+     *
+     * @param router the router for sending packets that were read.
+     * @param serverName the name of the server this socket is working for.
+     * @param connection the connection being read.
+     */
+    public StanzaHandler(PacketRouter router, String serverName, Connection connection) {
+        this.serverName = serverName;
+        this.router = router;
+        this.connection = connection;
+        // Create reader/parser
+        reader = new XMPPPacketReader();
+        reader.setXPPFactory(factory);
+    }
+
+    public void process(String stanza) throws Exception {
+
+        boolean initialStream = stanza.startsWith(STREAM_START);
+        if (!sessionCreated || initialStream) {
+            if (!initialStream) {
+                // Ignore <?xml version="1.0"?>
+                return;
+            }
+            // Found an stream:stream tag...
+            if (!sessionCreated) {
+                sessionCreated = true;
+                MXParser parser = (MXParser) factory.newPullParser();
+                parser.setInput(new StringReader(stanza));
+                createSession(parser);
+            } else if (startedSASL && session.getStatus() == Session.STATUS_AUTHENTICATED) {
+                startedSASL = false;
+                saslSuccessful();
+            } else if (waitingCompressionACK) {
+                waitingCompressionACK = false;
+                compressionSuccessful();
+            }
+            return;
+        }
+
+        // Create DOM object from received stanza
+        Element doc;
+        try {
+            doc = reader.read(new StringReader(stanza)).getRootElement();
+        } catch (DocumentException e) {
+            if (stanza.equals("</stream:stream>")) {
+                connection.close();
+                return;
+            }
+            // Throw the exception. This will close the connection
+            throw e;
+        }
+        if (doc == null) {
+            // No document found.
+            return;
+        }
+        String tag = doc.getName();
+        if ("starttls".equals(tag)) {
+            // Negotiate TLS
+            if (negotiateTLS()) {
+                tlsNegotiated();
+            } else {
+                connection.close();
+                session = null;
+            }
+        } else if ("auth".equals(tag)) {
+            // User is trying to authenticate using SASL
+            startedSASL = true;
+            // Forward packet to the server
+            process(doc);
+        } else if ("compress".equals(tag)) {
+            // Client is trying to initiate compression
+            if (compressClient(doc)) {
+                // Compression was successful so open a new stream and offer
+                // resource binding and session establishment (to client sessions only)
+                waitingCompressionACK = true;
+            }
+        } else {
+            process(doc);
+        }
+    }
+
+    private void process(Element doc) throws UnauthorizedException {
+        if (doc == null) {
+            return;
+        }
+
+        // Ensure that connection was secured if TLS was required
+        if (connection.getTlsPolicy() == Connection.TLSPolicy.required &&
+                !connection.isSecure()) {
+            closeNeverSecuredConnection();
+            return;
+        }
+
+        String tag = doc.getName();
+        if ("message".equals(tag)) {
+            Message packet;
+            try {
+                packet = new Message(doc, true);
+            }
+            catch(IllegalArgumentException e) {
+                Log.debug("Rejecting packet. JID malformed", e);
+                // The original packet contains a malformed JID so answer with an error.
+                Message reply = new Message();
+                reply.setID(doc.attributeValue("id"));
+                reply.setTo(session.getAddress());
+                reply.getElement().addAttribute("from", doc.attributeValue("to"));
+                reply.setError(PacketError.Condition.jid_malformed);
+                session.process(reply);
+                return;
+            }
+            processMessage(packet);
+        }
+        else if ("presence".equals(tag)) {
+            Presence packet;
+            try {
+                packet = new Presence(doc, true);
+            }
+            catch (IllegalArgumentException e) {
+                Log.debug("Rejecting packet. JID malformed", e);
+                // The original packet contains a malformed JID so answer an error
+                Presence reply = new Presence();
+                reply.setID(doc.attributeValue("id"));
+                reply.setTo(session.getAddress());
+                reply.getElement().addAttribute("from", doc.attributeValue("to"));
+                reply.setError(PacketError.Condition.jid_malformed);
+                session.process(reply);
+                return;
+            }
+            // Check that the presence type is valid. If not then assume available type
+            try {
+                packet.getType();
+            }
+            catch (IllegalArgumentException e) {
+                Log.warn("Invalid presence type", e);
+                // The presence packet contains an invalid presence type so replace it with
+                // an available presence type
+                packet.setType(null);
+            }
+            // Check that the presence show is valid. If not then assume available show value
+            try {
+                packet.getShow();
+            }
+            catch (IllegalArgumentException e) {
+                Log.warn("Invalid presence show", e);
+                // The presence packet contains an invalid presence show so replace it with
+                // an available presence show
+                packet.setShow(null);
+            }
+            if (session.getStatus() == Session.STATUS_CLOSED && packet.isAvailable()) {
+                // Ignore available presence packets sent from a closed session. A closed
+                // session may have buffered data pending to be processes so we want to ignore
+                // just Presences of type available
+                Log.warn("Ignoring available presence packet of closed session: " + packet);
+                return;
+            }
+            processPresence(packet);
+        }
+        else if ("iq".equals(tag)) {
+            IQ packet;
+            try {
+                packet = getIQ(doc);
+            }
+            catch(IllegalArgumentException e) {
+                Log.debug("Rejecting packet. JID malformed", e);
+                // The original packet contains a malformed JID so answer an error
+                IQ reply = new IQ();
+                if (!doc.elements().isEmpty()) {
+                    reply.setChildElement(((Element) doc.elements().get(0)).createCopy());
+                }
+                reply.setID(doc.attributeValue("id"));
+                reply.setTo(session.getAddress());
+                if (doc.attributeValue("to") != null) {
+                    reply.getElement().addAttribute("from", doc.attributeValue("to"));
+                }
+                reply.setError(PacketError.Condition.jid_malformed);
+                session.process(reply);
+                return;
+            }
+            processIQ(packet);
+        }
+        else
+        {
+            if (!processUnknowPacket(doc)) {
+                Log.warn(LocaleUtils.getLocalizedString("admin.error.packet.tag") +
+                        doc.asXML());
+                session.getConnection().close();
+            }
+        }
+    }
+
+    private IQ getIQ(Element doc) {
+        Element query = doc.element("query");
+        if (query != null && "jabber:iq:roster".equals(query.getNamespaceURI())) {
+            return new Roster(doc);
+        }
+        else {
+            return new IQ(doc, true);
+        }
+    }
+
+    /**
+     * Process the received IQ packet. Registered
+     * {@link org.jivesoftware.wildfire.interceptor.PacketInterceptor} will be invoked before
+     * and after the packet was routed.<p>
+     *
+     * Subclasses may redefine this method for different reasons such as modifying the sender
+     * of the packet to avoid spoofing, rejecting the packet or even process the packet in
+     * another thread.
+     *
+     * @param packet the received packet.
+     * @throws org.jivesoftware.wildfire.auth.UnauthorizedException if service is not available to sender.
+     */
+    protected void processIQ(IQ packet) throws UnauthorizedException {
+        router.route(packet);
+        session.incrementClientPacketCount();
+    }
+
+    /**
+     * Process the received Presence packet. Registered
+     * {@link org.jivesoftware.wildfire.interceptor.PacketInterceptor} will be invoked before
+     * and after the packet was routed.<p>
+     *
+     * Subclasses may redefine this method for different reasons such as modifying the sender
+     * of the packet to avoid spoofing, rejecting the packet or even process the packet in
+     * another thread.
+     *
+     * @param packet the received packet.
+     * @throws org.jivesoftware.wildfire.auth.UnauthorizedException if service is not available to sender.
+     */
+    protected void processPresence(Presence packet) throws UnauthorizedException {
+        router.route(packet);
+        session.incrementClientPacketCount();
+    }
+
+    /**
+     * Process the received Message packet. Registered
+     * {@link org.jivesoftware.wildfire.interceptor.PacketInterceptor} will be invoked before
+     * and after the packet was routed.<p>
+     *
+     * Subclasses may redefine this method for different reasons such as modifying the sender
+     * of the packet to avoid spoofing, rejecting the packet or even process the packet in
+     * another thread.
+     *
+     * @param packet the received packet.
+     * @throws org.jivesoftware.wildfire.auth.UnauthorizedException if service is not available to sender.
+     */
+    protected void processMessage(Message packet) throws UnauthorizedException {
+        router.route(packet);
+        session.incrementClientPacketCount();
+    }
+
+    /**
+     * Returns true if a received packet of an unkown type (i.e. not a Message, Presence
+     * or IQ) has been processed. If the packet was not processed then an exception will
+     * be thrown which will make the thread to stop processing further packets.
+     *
+     * @param doc the DOM element of an unkown type.
+     * @return  true if a received packet has been processed.
+     */
+    abstract boolean processUnknowPacket(Element doc);
+
+    /**
+     * Tries to secure the connection using TLS. If the connection is secured then reset
+     * the parser to use the new secured reader. But if the connection failed to be secured
+     * then send a <failure> stanza and close the connection.
+     *
+     * @return true if the connection was secured.
+     */
+    private boolean negotiateTLS() {
+        if (connection.getTlsPolicy() == Connection.TLSPolicy.disabled) {
+            // Set the not_authorized error
+            StreamError error = new StreamError(StreamError.Condition.not_authorized);
+            // Deliver stanza
+            connection.deliverRawText(error.toXML());
+            // Close the underlying connection
+            connection.close();
+            // Log a warning so that admins can track this case from the server side
+            Log.warn("TLS requested by initiator when TLS was never offered by server. " +
+                    "Closing connection : " + connection);
+            return false;
+        }
+        // Client requested to secure the connection using TLS. Negotiate TLS.
+        try {
+            connection.startTLS(false, null);
+        }
+        catch (Exception e) {
+            Log.error("Error while negotiating TLS", e);
+            connection.deliverRawText("<failure xmlns=\"urn:ietf:params:xml:ns:xmpp-tls\">");
+            connection.close();
+            return false;
+        }
+        return true;
+    }
+
+    /**
+     * TLS negotiation was successful so open a new stream and offer the new stream features.
+     * The new stream features will include available SASL mechanisms and specific features
+     * depending on the session type such as auth for Non-SASL authentication and register
+     * for in-band registration.
+     */
+    private void tlsNegotiated() {
+        // Offer stream features including SASL Mechanisms
+        StringBuilder sb = new StringBuilder(620);
+        sb.append(geStreamHeader());
+        sb.append("<stream:features>");
+        // Include available SASL Mechanisms
+        sb.append(SASLAuthentication.getSASLMechanisms(session));
+        // Include specific features such as auth and register for client sessions
+        String specificFeatures = session.getAvailableStreamFeatures();
+        if (specificFeatures != null) {
+            sb.append(specificFeatures);
+        }
+        sb.append("</stream:features>");
+        connection.deliverRawText(sb.toString());
+    }
+
+    /**
+     * After SASL authentication was successful we should open a new stream and offer
+     * new stream features such as resource binding and session establishment. Notice that
+     * resource binding and session establishment should only be offered to clients (i.e. not
+     * to servers or external components)
+     */
+    private void saslSuccessful() {
+        StringBuilder sb = new StringBuilder(420);
+        sb.append(geStreamHeader());
+        sb.append("<stream:features>");
+
+        // Include specific features such as resource binding and session establishment
+        // for client sessions
+        String specificFeatures = session.getAvailableStreamFeatures();
+        if (specificFeatures != null) {
+            sb.append(specificFeatures);
+        }
+        sb.append("</stream:features>");
+        connection.deliverRawText(sb.toString());
+    }
+
+    /**
+     * Start using compression but first check if the connection can and should use compression.
+     * The connection will be closed if the requested method is not supported, if the connection
+     * is already using compression or if client requested to use compression but this feature
+     * is disabled.
+     *
+     * @param doc the element sent by the client requesting compression. Compression method is
+     *            included.
+     * @return true if it was possible to use compression.
+     */
+    private boolean compressClient(Element doc) {
+        String error = null;
+        if (connection.getCompressionPolicy() == Connection.CompressionPolicy.disabled) {
+            // Client requested compression but this feature is disabled
+            error = "<failure xmlns='http://jabber.org/protocol/compress'><setup-failed/></failure>";
+            // Log a warning so that admins can track this case from the server side
+            Log.warn("Client requested compression while compression is disabled. Closing " +
+                    "connection : " + connection);
+        } else if (connection.isCompressed()) {
+            // Client requested compression but connection is already compressed
+            error = "<failure xmlns='http://jabber.org/protocol/compress'><setup-failed/></failure>";
+            // Log a warning so that admins can track this case from the server side
+            Log.warn("Client requested compression and connection is already compressed. Closing " +
+                    "connection : " + connection);
+        } else {
+            // Check that the requested method is supported
+            String method = doc.elementText("method");
+            if (!"zlib".equals(method)) {
+                error = "<failure xmlns='http://jabber.org/protocol/compress'><unsupported-method/></failure>";
+                // Log a warning so that admins can track this case from the server side
+                Log.warn("Requested compression method is not supported: " + method +
+                        ". Closing connection : " + connection);
+            }
+        }
+
+        if (error != null) {
+            // Deliver stanza
+            connection.deliverRawText(error);
+            return false;
+        } else {
+            // Indicate client that he can proceed and compress the socket
+            connection.deliverRawText("<compressed xmlns='http://jabber.org/protocol/compress'/>");
+
+            // Start using compression
+            connection.startCompression();
+            return true;
+        }
+    }
+
+    /**
+     * After compression was successful we should open a new stream and offer
+     * new stream features such as resource binding and session establishment. Notice that
+     * resource binding and session establishment should only be offered to clients (i.e. not
+     * to servers or external components)
+     */
+    private void compressionSuccessful() {
+        StringBuilder sb = new StringBuilder(340);
+        sb.append(geStreamHeader());
+        sb.append("<stream:features>");
+        // Include SASL mechanisms only if client has not been authenticated
+        if (session.getStatus() != Session.STATUS_AUTHENTICATED) {
+            // Include available SASL Mechanisms
+            sb.append(SASLAuthentication.getSASLMechanisms(session));
+        }
+        // Include specific features such as resource binding and session establishment
+        // for client sessions
+        String specificFeatures = session.getAvailableStreamFeatures();
+        if (specificFeatures != null) {
+            sb.append(specificFeatures);
+        }
+        sb.append("</stream:features>");
+        connection.deliverRawText(sb.toString());
+    }
+
+    private String geStreamHeader() {
+        StringBuilder sb = new StringBuilder(200);
+        sb.append("<?xml version='1.0' encoding='");
+        sb.append(CHARSET);
+        sb.append("'?>");
+        if (connection.isFlashClient()) {
+            sb.append("<flash:stream xmlns:flash=\"http://www.jabber.com/streams/flash\" ");
+        } else {
+            sb.append("<stream:stream ");
+        }
+        sb.append("xmlns:stream=\"http://etherx.jabber.org/streams\" xmlns=\"");
+        sb.append(getNamespace());
+        sb.append("\" from=\"");
+        sb.append(serverName);
+        sb.append("\" id=\"");
+        sb.append(session.getStreamID());
+        sb.append("\" xml:lang=\"");
+        sb.append(connection.getLanguage());
+        sb.append("\" version=\"");
+        sb.append(Session.MAJOR_VERSION).append(".").append(Session.MINOR_VERSION);
+        sb.append("\">");
+        return sb.toString();
+    }
+
+    /**
+     * Close the connection since TLS was mandatory and the entity never negotiated TLS. Before
+     * closing the connection a stream error will be sent to the entity.
+     */
+    void closeNeverSecuredConnection() {
+        // Set the not_authorized error
+        StreamError error = new StreamError(StreamError.Condition.not_authorized);
+        // Deliver stanza
+        connection.deliverRawText(error.toXML());
+        // Close the underlying connection
+        connection.close();
+        // Log a warning so that admins can track this case from the server side
+        Log.warn("TLS was required by the server and connection was never secured. " +
+                "Closing connection : " + connection);
+    }
+
+    /**
+     * Uses the XPP to grab the opening stream tag and create an active session
+     * object. The session to create will depend on the sent namespace. In all
+     * cases, the method obtains the opening stream tag, checks for errors, and
+     * either creates a session or returns an error and kills the connection.
+     * If the connection remains open, the XPP will be set to be ready for the
+     * first packet. A call to next() should result in an START_TAG state with
+     * the first packet in the stream.
+     */
+    protected void createSession(XmlPullParser xpp) throws XmlPullParserException, IOException {
+        for (int eventType = xpp.getEventType(); eventType != XmlPullParser.START_TAG;) {
+            eventType = xpp.next();
+        }
+
+        // Check that the TO attribute of the stream header matches the server name or a valid
+        // subdomain. If the value of the 'to' attribute is not valid then return a host-unknown
+        // error and close the underlying connection.
+        String host = xpp.getAttributeValue("", "to");
+        if (validateHost() && isHostUnknown(host)) {
+            StringBuilder sb = new StringBuilder(250);
+            sb.append("<?xml version='1.0' encoding='");
+            sb.append(CHARSET);
+            sb.append("'?>");
+            // Append stream header
+            sb.append("<stream:stream ");
+            sb.append("from=\"").append(serverName).append("\" ");
+            sb.append("id=\"").append(StringUtils.randomString(5)).append("\" ");
+            sb.append("xmlns=\"").append(xpp.getNamespace(null)).append("\" ");
+            sb.append("xmlns:stream=\"").append(xpp.getNamespace("stream")).append("\" ");
+            sb.append("version=\"1.0\">");
+            // Set the host_unknown error
+            StreamError error = new StreamError(StreamError.Condition.host_unknown);
+            sb.append(error.toXML());
+            // Deliver stanza
+            connection.deliverRawText(sb.toString());
+            // Close the underlying connection
+            connection.close();
+            // Log a warning so that admins can track this cases from the server side
+            Log.warn("Closing session due to incorrect hostname in stream header. Host: " + host +
+                    ". Connection: " + connection);
+        }
+
+        // Create the correct session based on the sent namespace. At this point the server
+        // may offer the client to secure the connection. If the client decides to secure
+        // the connection then a <starttls> stanza should be received
+        else if (!createSession(xpp.getNamespace(null), serverName, xpp, connection)) {
+            // No session was created because of an invalid namespace prefix so answer a stream
+            // error and close the underlying connection
+            StringBuilder sb = new StringBuilder(250);
+            sb.append("<?xml version='1.0' encoding='");
+            sb.append(CHARSET);
+            sb.append("'?>");
+            // Append stream header
+            sb.append("<stream:stream ");
+            sb.append("from=\"").append(serverName).append("\" ");
+            sb.append("id=\"").append(StringUtils.randomString(5)).append("\" ");
+            sb.append("xmlns=\"").append(xpp.getNamespace(null)).append("\" ");
+            sb.append("xmlns:stream=\"").append(xpp.getNamespace("stream")).append("\" ");
+            sb.append("version=\"1.0\">");
+            // Include the bad-namespace-prefix in the response
+            StreamError error = new StreamError(StreamError.Condition.bad_namespace_prefix);
+            sb.append(error.toXML());
+            connection.deliverRawText(sb.toString());
+            // Close the underlying connection
+            connection.close();
+            // Log a warning so that admins can track this cases from the server side
+            Log.warn("Closing session due to bad_namespace_prefix in stream header. Prefix: " +
+                    xpp.getNamespace(null) + ". Connection: " + connection);
+        }
+    }
+
+    private boolean isHostUnknown(String host) {
+        if (host == null) {
+            // Answer false since when using server dialback the stream header will not
+            // have a TO attribute
+            return false;
+        }
+        if (serverName.equals(host)) {
+            // requested host matched the server name
+            return false;
+        }
+        return true;
+    }
+
+    /**
+     * Returns the stream namespace. (E.g. jabber:client, jabber:server, etc.).
+     *
+     * @return the stream namespace.
+     */
+    abstract String getNamespace();
+
+    /**
+     * Returns true if the value of the 'to' attribute in the stream header should be
+     * validated. If the value of the 'to' attribute is not valid then a host-unknown error
+     * will be returned and the underlying connection will be closed.
+     *
+     * @return true if the value of the 'to' attribute in the initial stream header should be
+     *         validated.
+     */
+    abstract boolean validateHost();
+
+    /**
+     * Creates the appropriate {@link Session} subclass based on the specified namespace.
+     *
+     * @param namespace the namespace sent in the stream element. eg. jabber:client.
+     * @return the created session or null.
+     * @throws org.xmlpull.v1.XmlPullParserException
+     */
+    abstract boolean createSession(String namespace, String serverName, XmlPullParser xpp, Connection connection)
+            throws XmlPullParserException;
+}

src/java/org/jivesoftware/wildfire/nio/ByteBufferWriter.java

+/**
+ * $Revision: $
+ * $Date: $
+ *
+ * Copyright (C) 2007 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.nio;
+
+import org.apache.mina.common.ByteBuffer;
+
+import java.io.IOException;
+import java.io.Writer;
+import java.nio.charset.CharsetEncoder;
+
+/**
+ * Wrapper on a MINA {@link ByteBuffer} that extends the Writer class.
+ *
+ * @author Gaston Dombia
+ */
+public class ByteBufferWriter extends Writer {
+    private CharsetEncoder encoder;
+    private ByteBuffer byteBuffer;
+
+
+    public ByteBufferWriter(ByteBuffer byteBuffer, CharsetEncoder encoder) {
+        this.encoder = encoder;
+        this.byteBuffer = byteBuffer;
+    }
+
+    public void write(char cbuf[], int off, int len) throws IOException {
+        byteBuffer.putString(new String(cbuf, off, len), encoder);
+    }
+
+    public void flush() throws IOException {
+        // Ignore
+    }
+
+    public void close() throws IOException {
+        // Ignore
+    }
+}

src/java/org/jivesoftware/wildfire/nio/ConnectionHandler.java

+/**
+ * $Revision: $
+ * $Date: $
+ *
+ * Copyright (C) 2007 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.nio;
+
+import org.apache.mina.common.IdleStatus;
+import org.apache.mina.common.IoHandlerAdapter;
+import org.apache.mina.common.IoSession;
+import org.jivesoftware.util.Log;
+import org.jivesoftware.wildfire.Connection;
+import org.jivesoftware.wildfire.net.StanzaHandler;
+
+import java.io.IOException;
+
+/**
+ * A ConnectionHandler is responsible for creating new sessions, destroying sessions and delivering
+ * received XML stanzas to the proper StanzaHandler.
+ *
+ * @author Gaston Dombiak
+ */
+public abstract class ConnectionHandler extends IoHandlerAdapter {
+
+    /**
+     * The utf-8 charset for decoding and encoding Jabber packet streams.
+     */
+    static final String CHARSET = "UTF-8";
+    static final String XML_PARSER = "XML-PARSER";
+    private static final String HANDLER = "HANDLER";
+    private static final String CONNECTION = "CONNECTION";
+
+    protected String serverName;
+
+
+    protected ConnectionHandler(String serverName) {
+        this.serverName = serverName;
+    }
+
+    public void sessionOpened(IoSession session) throws Exception {
+        // Create a new XML parser for the new connection. The parser will be used by the XMPPDecoder filter.
+        XMLLightweightParser parser = new XMLLightweightParser(CHARSET);
+        session.setAttribute(XML_PARSER, parser);
+        // Create a new NIOConnection for the new session
+        NIOConnection connection = createNIOConnection(session);
+        session.setAttribute(CONNECTION, connection);
+        session.setAttribute(HANDLER, createStanzaHandler(connection));
+        // Set the max time a connection can be idle before closing it
+        int idleTime = getMaxIdleTime();
+        if (idleTime > 0) {
+            session.setIdleTime(IdleStatus.BOTH_IDLE, idleTime);
+        }
+    }
+
+    public void sessionClosed(IoSession session) throws Exception {
+        // Get the connection for this session
+        Connection connection = (Connection) session.getAttribute(CONNECTION);
+        // Inform the connection that it was closed
+        connection.close();
+    }
+
+    public void sessionIdle(IoSession session, IdleStatus status) throws Exception {
+        // Get the connection for this session
+        Connection connection = (Connection) session.getAttribute(CONNECTION);
+        // Close idle connection
+        if (Log.isDebugEnabled()) {
+            Log.debug("Closing connection that has been idle: " + connection);
+        }
+        connection.close();
+    }
+
+    public void exceptionCaught(IoSession session, Throwable cause) throws Exception {
+        if (cause instanceof IOException) {
+            // TODO Verify if there were packets pending to be sent and decide what to do with them
+            Log.debug(cause);
+        }
+        else {
+            Log.error(cause);
+        }
+    }
+
+    public void messageReceived(IoSession session, Object message) throws Exception {
+        //System.out.println("RCVD: " + message);
+        // Get the stanza handler for this session
+        StanzaHandler handler = (StanzaHandler) session.getAttribute(HANDLER);
+        // Let the stanza handler process the received stanza
+        try {
+            handler.process( (String) message);
+        } catch (Exception e) {
+            Log.error("Closing connection due to error while processing message: " + message, e);
+            Connection connection = (Connection) session.getAttribute(CONNECTION);
+            connection.close();
+        }
+    }
+
+    abstract NIOConnection createNIOConnection(IoSession session);
+
+    abstract StanzaHandler createStanzaHandler(NIOConnection connection);
+
+    /**
+     * Returns the max number of seconds a connection can be idle (both ways) before
+     * being closed.<p>
+     *
+     * @return the max number of seconds a connection can be idle.
+     */
+    abstract int getMaxIdleTime();
+}

src/java/org/jivesoftware/wildfire/nio/MultiplexerConnectionHandler.java

+/**
+ * $Revision: $
+ * $Date: $
+ *
+ * Copyright (C) 2007 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.nio;
+
+import org.apache.mina.common.IoSession;
+import org.jivesoftware.util.JiveGlobals;
+import org.jivesoftware.wildfire.XMPPServer;
+import org.jivesoftware.wildfire.multiplex.MultiplexerPacketDeliverer;
+import org.jivesoftware.wildfire.net.MultiplexerStanzaHandler;
+import org.jivesoftware.wildfire.net.StanzaHandler;
+
+/**
+ * ConnectionHandler that knows which subclass of {@link org.jivesoftware.wildfire.net.StanzaHandler} should
+ * be created and how to build and configure a {@link org.jivesoftware.wildfire.nio.NIOConnection}.
+ *
+ * @author Gaston Dombiak
+ */
+public class MultiplexerConnectionHandler extends ConnectionHandler {
+
+    public MultiplexerConnectionHandler(String serverName) {
+        super(serverName);
+    }
+
+    NIOConnection createNIOConnection(IoSession session) {
+        return new NIOConnection(session, new MultiplexerPacketDeliverer());
+    }
+
+    StanzaHandler createStanzaHandler(NIOConnection connection) {
+        return new MultiplexerStanzaHandler(XMPPServer.getInstance().getPacketRouter(), serverName, connection);
+    }
+
+    int getMaxIdleTime() {
+        return JiveGlobals.getIntProperty("xmpp.multiplex.idle", 5 * 60 * 1000) / 1000;
+    }
+}

src/java/org/jivesoftware/wildfire/nio/NIOConnection.java

+/**
+ * $Revision: $
+ * $Date: $
+ *
+ * Copyright (C) 2007 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.nio;
+
+import org.apache.mina.common.ByteBuffer;
+import org.apache.mina.common.IoFilterChain;
+import org.apache.mina.common.IoSession;
+import org.apache.mina.filter.CompressionFilter;
+import org.apache.mina.filter.SSLFilter;
+import org.dom4j.io.OutputFormat;
+import org.jivesoftware.util.JiveGlobals;
+import org.jivesoftware.util.Log;
+import org.jivesoftware.util.XMLWriter;
+import org.jivesoftware.wildfire.Connection;
+import org.jivesoftware.wildfire.ConnectionCloseListener;
+import org.jivesoftware.wildfire.PacketDeliverer;
+import org.jivesoftware.wildfire.auth.UnauthorizedException;
+import org.jivesoftware.wildfire.net.SSLConfig;
+import org.jivesoftware.wildfire.net.SSLJiveKeyManagerFactory;
+import org.jivesoftware.wildfire.net.SSLJiveTrustManagerFactory;
+import org.jivesoftware.wildfire.net.ServerTrustManager;
+import org.jivesoftware.wildfire.session.Session;
+import org.xmpp.packet.Packet;
+
+import javax.net.ssl.KeyManager;
+import javax.net.ssl.SSLContext;
+import javax.net.ssl.TrustManager;
+import java.net.InetAddress;
+import java.net.InetSocketAddress;
+import java.net.UnknownHostException;
+import java.nio.charset.Charset;
+import java.nio.charset.CharsetEncoder;
+import java.security.KeyStore;
+import java.util.HashMap;
+import java.util.Map;
+
+/**
+ * Implementation of {@link Connection} inteface specific for NIO connections when using
+ * the MINA framework.<p>
+ *
+ * MINA project can be found at <a href="http://mina.apache.org">here</a>.
+ *
+ * @author Gaston Dombiak
+ */
+public class NIOConnection implements Connection {
+
+    /**
+     * The utf-8 charset for decoding and encoding XMPP packet streams.
+     */
+    public static final String CHARSET = "UTF-8";
+
+    private Session session;
+    private IoSession ioSession;
+
+    final private Map<ConnectionCloseListener, Object> listeners =
+            new HashMap<ConnectionCloseListener, Object>();
+    
+    /**
+     * Deliverer to use when the connection is closed or was closed when delivering
+     * a packet.
+     */
+    private PacketDeliverer backupDeliverer;
+    private boolean flashClient = false;
+    private int majorVersion = 1;
+    private int minorVersion = 0;
+    private String language = null;
+
+    // TODO Uso el #checkHealth????
+    /**
+     * TLS policy currently in use for this connection.
+     */
+    private TLSPolicy tlsPolicy = TLSPolicy.optional;
+
+    /**
+     * Compression policy currently in use for this connection.
+     */
+    private CompressionPolicy compressionPolicy = CompressionPolicy.disabled;
+    private CharsetEncoder encoder;
+
+
+    public NIOConnection(IoSession session, PacketDeliverer packetDeliverer) {
+        this.ioSession = session;
+        this.backupDeliverer = packetDeliverer;
+        encoder = Charset.forName(CHARSET).newEncoder();
+    }
+
+    public boolean validate() {
+        if (isClosed()) {
+            return false;
+        }
+        deliverRawText(" ");
+        return !isClosed();
+    }
+
+    public void registerCloseListener(ConnectionCloseListener listener, Object handbackMessage) {
+        if (isClosed()) {
+            listener.onConnectionClose(handbackMessage);
+        }
+        else {
+            listeners.put(listener, handbackMessage);
+        }
+    }
+
+    public void removeCloseListener(ConnectionCloseListener listener) {
+        listeners.remove(listener);
+    }
+
+    public InetAddress getInetAddress() throws UnknownHostException {
+        return ((InetSocketAddress) ioSession.getRemoteAddress()).getAddress();
+    }
+
+    public PacketDeliverer getPacketDeliverer() {
+        return backupDeliverer;
+    }
+
+    public void close() {
+        boolean wasClosed = false;
+        synchronized (this) {
+            if (!isClosed()) {
+                try {
+                    deliverRawText("</stream:stream>");
+                } catch (Exception e) {
+                    // Ignore
+                }
+                closeConnection();
+                wasClosed = true;
+            }
+        }
+        if (wasClosed) {
+            notifyCloseListeners();
+        }
+    }
+
+    public void systemShutdown() {
+        deliverRawText("<stream:error><system-shutdown " +
+                "xmlns='urn:ietf:params:xml:ns:xmpp-streams'/></stream:error>");
+        close();
+    }
+
+    /**
+     * Forces the connection to be closed immediately no matter if closing the socket takes
+     * a long time. This method should only be called from {@link org.jivesoftware.wildfire.net.SocketSendingTracker} when
+     * sending data over the socket has taken a long time and we need to close the socket, discard
+     * the connection and its ioSession.
+     */
+    private void forceClose() {
+        closeConnection();
+        // Notify the close listeners so that the SessionManager can send unavailable
+        // presences if required.
+        notifyCloseListeners();
+    }
+
+    private void closeConnection() {
+        ioSession.close();
+    }
+
+    /**
+     * Notifies all close listeners that the connection has been closed.
+     * Used by subclasses to properly finish closing the connection.
+     */
+    private void notifyCloseListeners() {
+        synchronized (listeners) {
+            for (ConnectionCloseListener listener : listeners.keySet()) {
+                try {
+                    listener.onConnectionClose(listeners.get(listener));
+                }
+                catch (Exception e) {
+                    Log.error("Error notifying listener: " + listener, e);
+                }
+            }
+        }
+    }
+
+    public void init(Session owner) {
+        session = owner;
+    }
+
+    public boolean isClosed() {
+        if (session == null) {
+            return !ioSession.isConnected();
+        }
+        return session.getStatus() == Session.STATUS_CLOSED;
+    }
+
+    public boolean isSecure() {
+        return ioSession.getFilterChain().contains("tls");
+    }
+
+    public void deliver(Packet packet) throws UnauthorizedException {
+        if (isClosed()) {
+            backupDeliverer.deliver(packet);
+        }
+        else {
+            ByteBuffer buffer = ByteBuffer.allocate(4096);
+            buffer.setAutoExpand(true);
+
+            boolean errorDelivering = false;
+            try {
+                //XMLWriter xmlSerializer = new XMLWriter(buffer.asOutputStream(), new OutputFormat());
+                XMLWriter xmlSerializer = new XMLWriter(new ByteBufferWriter(buffer, encoder), new OutputFormat());
+                xmlSerializer.write(packet.getElement());
+                xmlSerializer.flush();
+                if (flashClient) {
+                    buffer.put((byte) '\0');
+                }
+                buffer.flip();
+                //System.out.println("SENT: " + doc.asXML());
+                ioSession.write(buffer);
+            }
+            catch (Exception e) {
+                Log.debug("Error delivering packet" + "\n" + this.toString(), e);
+                errorDelivering = true;
+            }
+            if (errorDelivering) {
+                close();
+                // Retry sending the packet again. Most probably if the packet is a
+                // Message it will be stored offline
+                backupDeliverer.deliver(packet);
+            }
+        }
+    }
+
+    public void deliverRawText(String text) {
+        if (!isClosed()) {
+            ByteBuffer buffer = ByteBuffer.allocate(text.length());
+            buffer.setAutoExpand(true);
+
+            boolean errorDelivering = false;
+            try {
+                //Charset charset = Charset.forName(CHARSET);
+                //buffer.putString(text, charset.newEncoder());
+                buffer.put(text.getBytes(CHARSET));
+                if (flashClient) {
+                    buffer.put((byte) '\0');
+                }
+                buffer.flip();
+                //System.out.println("SENT: " + text);
+                ioSession.write(buffer);
+            }
+            catch (Exception e) {
+                Log.debug("Error delivering raw text" + "\n" + this.toString(), e);
+                errorDelivering = true;
+            }
+            if (errorDelivering) {
+                close();
+            }
+        }
+    }
+
+    public void startTLS(boolean clientMode, String remoteServer) throws Exception {
+        KeyStore ksKeys = SSLConfig.getKeyStore();
+        String keypass = SSLConfig.getKeyPassword();
+
+        KeyStore ksTrust = SSLConfig.getTrustStore();
+        String trustpass = SSLConfig.getTrustPassword();
+
+        // KeyManager's decide which key material to use.
+        KeyManager[] km = SSLJiveKeyManagerFactory.getKeyManagers(ksKeys, keypass);
+
+        // TrustManager's decide whether to allow connections.
+        TrustManager[] tm = SSLJiveTrustManagerFactory.getTrustManagers(ksTrust, trustpass);
+        // TODO Set proper value when s2s is supported
+        boolean needClientAuth = false;
+        if (clientMode || needClientAuth) {
+            // Check if we can trust certificates presented by the server
+            tm = new TrustManager[]{new ServerTrustManager(remoteServer, ksTrust)};
+        }
+
+        SSLContext tlsContext = SSLContext.getInstance("TLS");
+
+        tlsContext.init(km, tm, null);
+
+        SSLFilter filter = new SSLFilter(tlsContext);
+        filter.setUseClientMode(clientMode);
+        if (needClientAuth) {
+            // Only REQUIRE client authentication if we are fully verifying certificates
+            if (JiveGlobals.getBooleanProperty("xmpp.server.certificate.verify", true) &&
+                    JiveGlobals.getBooleanProperty("xmpp.server.certificate.verify.chain", true) &&
+                    !JiveGlobals
+                            .getBooleanProperty("xmpp.server.certificate.accept-selfsigned", false))
+            {
+                filter.setNeedClientAuth(true);
+            }
+            else {
+                // Just indicate that we would like to authenticate the client but if client
+                // certificates are self-signed or have no certificate chain then we are still
+                // good
+                filter.setWantClientAuth(true);
+            }
+        }
+
+        ioSession.getFilterChain().addAfter("org.apache.mina.common.ExecutorThreadModel", "tls", filter);
+        ioSession.setAttribute(SSLFilter.DISABLE_ENCRYPTION_ONCE, Boolean.TRUE);
+        if (!clientMode) {
+            // Indicate the client that the server is ready to negotiate TLS
+            deliverRawText("<proceed xmlns=\"urn:ietf:params:xml:ns:xmpp-tls\"/>");
+        }
+    }
+
+    public void startCompression() {
+        IoFilterChain chain = ioSession.getFilterChain();
+        String baseFilter = "org.apache.mina.common.ExecutorThreadModel";
+        if (chain.contains("tls")) {
+            baseFilter = "tls";
+        }
+        chain.addAfter(baseFilter, "compression", new CompressionFilter(CompressionFilter.COMPRESSION_MAX));
+    }
+
+    public boolean isFlashClient() {
+        return flashClient;
+    }
+
+    public void setFlashClient(boolean flashClient) {
+        this.flashClient = flashClient;
+    }
+
+    public int getMajorXMPPVersion() {
+        return majorVersion;
+    }
+
+    public int getMinorXMPPVersion() {
+        return minorVersion;
+    }
+
+    public void setXMPPVersion(int majorVersion, int minorVersion) {
+        this.majorVersion = majorVersion;
+        this.minorVersion = minorVersion;
+    }
+
+    public String getLanguage() {
+        return language;
+    }
+
+    public void setLanaguage(String language) {
+        this.language = language;
+    }
+
+    public boolean isCompressed() {
+        return ioSession.getFilterChain().contains("compression");
+    }
+
+    public CompressionPolicy getCompressionPolicy() {
+        return compressionPolicy;
+    }
+
+    public void setCompressionPolicy(CompressionPolicy compressionPolicy) {
+        this.compressionPolicy = compressionPolicy;
+    }
+
+    public TLSPolicy getTlsPolicy() {
+        return tlsPolicy;
+    }
+
+    public void setTlsPolicy(TLSPolicy tlsPolicy) {
+        this.tlsPolicy = tlsPolicy;
+    }
+
+    public String toString() {
+        return super.toString() + " MINA Session: " + ioSession;
+    }
+}

src/java/org/jivesoftware/wildfire/nio/XMLLightweightParser.java

+/**
+ * $Revision: $
+ * $Date: $
+ *
+ * Copyright (C) 2007 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.nio;
+
+import org.apache.mina.common.ByteBuffer;
+
+import java.nio.CharBuffer;
+import java.nio.charset.Charset;
+import java.util.ArrayList;
+import java.util.List;
+
+/**
+ * This is a Light-Weight XML Parser.
+ * It read data from a channel and collect data until data are available in
+ * the channel.
+ * When a message is complete you can retrieve messages invoking the method
+ * getMsgs() and you can invoke the method areThereMsgs() to know if at least
+ * an message is presents.
+ *
+ * @author Daniele Piras
+ */
+class XMLLightweightParser {
+    // Chars that rappresent CDATA section start
+    protected static char[] CDATA_START = {'<', '!', '[', 'C', 'D', 'A', 'T', 'A', '['};
+    // Chars that rappresent CDATA section end
+    protected static char[] CDATA_END = {']', ']', '>'};
+
+    // Buffer with all data retrieved
+    protected StringBuilder buffer = new StringBuilder();
+
+    // ---- INTERNAL STATUS -------
+    // Initial status
+    protected static final int INIT = 0;
+    // Status used when the first tag name is retrieved
+    protected static final int HEAD = 2;
+    // Status used when robot is inside the xml and it looking for the tag conclusion
+    protected static final int INSIDE = 3;
+    // Status used when a '<' is found and try to find the conclusion tag.
+    protected static final int PRETAIL = 4;
+    // Status used when the ending tag is equal to the head tag
+    protected static final int TAIL = 5;
+    // Status used when robot is inside the main tag and found an '/' to check '/>'.
+    protected static final int VERIFY_CLOSE_TAG = 6;
+    //  Status used when you are inside a parameter
+    protected static final int INSIDE_PARAM_VALUE = 7;
+    //  Status used when you are inside a cdata section
+    protected static final int INSIDE_CDATA = 8;
+
+
+    // Current robot status
+    protected int status = XMLLightweightParser.INIT;
+
+    // Index to looking for a CDATA section start or end.
+    protected int cdataOffset = 0;
+
+    // Number of chars that machs with the head tag. If the tailCount is equal to
+    // the head length so a close tag is found.
+    protected int tailCount = 0;
+    // Indicate the starting point in the buffer for the next message.
+    protected int startLastMsg = 0;
+    // Flag used to discover tag in the form <tag />.
+    protected boolean insideRootTag = false;
+    // Object conteining the head tag
+    protected StringBuilder head = new StringBuilder(5);
+    // List with all finished messages found.
+    protected List<String> msgs = new ArrayList<String>();
+
+    protected boolean insideChildrenTag = false;
+
+    ByteBuffer byteBuffer;
+    Charset encoder;
+
+    public XMLLightweightParser(String charset) {
+        encoder = Charset.forName(charset);
+    }
+
+    /*
+    * true if the parser has found some complete xml message.
+    */
+    public boolean areThereMsgs() {
+        return (msgs.size() > 0);
+    }
+
+    /*
+    * @return an array with all messages found
+    */
+    public String[] getMsgs() {
+        String[] res = new String[msgs.size()];
+        for (int i = 0; i < res.length; i++) {
+            res[i] = msgs.get(i);
+        }
+        msgs.clear();
+        invalidateBuffer();
+        return res;
+    }
+
+    /*
+    * Method use to re-initialize the buffer
+    */
+    protected void invalidateBuffer() {
+        if (buffer.length() > 0) {
+            String str = buffer.substring(startLastMsg);
+            buffer.delete(0, buffer.length());
+            buffer.append(str);
+            buffer.trimToSize();
+        }
+        startLastMsg = 0;
+    }
+
+
+    /*
+    * Method that add a message to the list and reinit parser.
+    */
+    protected void foundMsg(String msg) {
+        // Add message to the complete message list
+        if (msg != null) {
+            msgs.add(msg);
+        }
+        // Move the position into the buffer
+        status = XMLLightweightParser.INIT;
+        tailCount = 0;
+        cdataOffset = 0;
+        head.setLength(0);
+        insideRootTag = false;
+        insideChildrenTag = false;
+    }
+
+    /*
+    * Main reading method
+    */
+    public void read(ByteBuffer byteBuffer) throws Exception {
+        int readByte = byteBuffer.remaining();
+
+        invalidateBuffer();
+        CharBuffer charBuffer = encoder.decode(byteBuffer.buf());
+        //charBuffer.flip();
+        char[] buf = charBuffer.array();
+
+        buffer.append(buf);
+        // Robot.
+        char ch;
+        for (int i = 0; i < readByte; i++) {
+            //ch = rawByteBuffer[ i ];
+            ch = buf[i];
+            if (status == XMLLightweightParser.TAIL) {
+                // Looking for the close tag
+                if (ch == head.charAt(tailCount)) {
+                    tailCount++;
+                    if (tailCount == head.length()) {
+                        // Close tag found!
+                        // Calculate the correct start,end position of the message into the buffer
+                        int end = buffer.length() - readByte + (i + 1);
+                        String msg = buffer.substring(startLastMsg, end);
+                        // Add message to the list
+                        foundMsg(msg);
+                        startLastMsg = end;
+                    }
+                } else {
+                    tailCount = 0;
+                    status = XMLLightweightParser.INSIDE;
+                }
+            } else if (status == XMLLightweightParser.PRETAIL) {
+                if (ch == XMLLightweightParser.CDATA_START[cdataOffset]) {
+                    cdataOffset++;
+                    if (cdataOffset == XMLLightweightParser.CDATA_START.length) {
+                        status = XMLLightweightParser.INSIDE_CDATA;
+                        cdataOffset = 0;
+                        continue;
+                    }
+                } else {
+                    cdataOffset = 0;
+                    status = XMLLightweightParser.INSIDE;
+                }
+                if (ch == '/') {
+                    status = XMLLightweightParser.TAIL;
+                }
+            } else if (status == XMLLightweightParser.VERIFY_CLOSE_TAG) {
+                if (ch == '>') {
+                    // Found a tag in the form <tag />
+                    int end = buffer.length() - readByte + (i + 1);
+                    String msg = buffer.substring(startLastMsg, end);
+                    // Add message to the list
+                    foundMsg(msg);
+                    startLastMsg = end;
+                } else {
+                    status = XMLLightweightParser.INSIDE;
+                }
+            } else if (status == XMLLightweightParser.INSIDE_PARAM_VALUE) {
+
+                if (ch == '"') {
+                    status = XMLLightweightParser.INSIDE;
+                    continue;
+                }
+            } else if (status == XMLLightweightParser.INSIDE_CDATA) {
+                if (ch == XMLLightweightParser.CDATA_END[cdataOffset]) {
+                    cdataOffset++;
+                    if (cdataOffset == XMLLightweightParser.CDATA_END.length) {
+                        status = XMLLightweightParser.INSIDE;
+                        cdataOffset = 0;
+                        continue;
+                    }
+                } else {
+                    cdataOffset = 0;
+                }
+            } else if (status == XMLLightweightParser.INSIDE) {
+                if (ch == XMLLightweightParser.CDATA_START[cdataOffset]) {
+                    cdataOffset++;
+                    if (cdataOffset == XMLLightweightParser.CDATA_START.length) {
+                        status = XMLLightweightParser.INSIDE_CDATA;
+                        cdataOffset = 0;
+                        continue;
+                    }
+                } else {
+                    cdataOffset = 0;
+                }
+                if (ch == '"') {
+                    status = XMLLightweightParser.INSIDE_PARAM_VALUE;
+                } else if (ch == '>') {
+                    if (insideRootTag &&
+                            ("stream:stream>".equals(head.toString()) || ("?xml>".equals(head.toString())))) {
+                        // Found closing stream:stream
+                        int end = buffer.length() - readByte + (i + 1);
+                        String msg = buffer.substring(startLastMsg, end);
+                        foundMsg(msg);
+                        startLastMsg = end;
+                    }
+                    insideRootTag = false;
+                } else if (ch == '<') {
+                    status = XMLLightweightParser.PRETAIL;
+                    insideChildrenTag = true;
+                } else if (ch == '/' && insideRootTag && !insideChildrenTag) {
+                    status = XMLLightweightParser.VERIFY_CLOSE_TAG;
+                }
+            } else if (status == XMLLightweightParser.HEAD) {
+                if (ch == ' ' || ch == '>') {
+                    // Append > to head to facility the research of </tag>
+                    head.append(">");
+                    status = XMLLightweightParser.INSIDE;
+                    insideRootTag = true;
+                    insideChildrenTag = false;
+                    continue;
+                }
+                head.append(ch);
+
+            } else if (status == XMLLightweightParser.INIT) {
+                if (ch != ' ' && ch != '\r' && ch != '\n' && ch != '<') {
+                    invalidateBuffer();
+                    return;
+                }
+                if (ch == '<') {
+                    status = XMLLightweightParser.HEAD;
+                }
+            }
+        }
+        if (head.length() > 0 && "/stream:stream>".equals(head.toString())) {
+            // Found closing stream:stream
+            foundMsg("</stream:stream>");
+        }
+    }
+}

src/java/org/jivesoftware/wildfire/nio/XMPPCodecFactory.java

+/**
+ * $Revision: $
+ * $Date: $
+ *
+ * Copyright (C) 2007 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.nio;
+
+import org.apache.mina.filter.codec.ProtocolCodecFactory;
+import org.apache.mina.filter.codec.ProtocolDecoder;
+import org.apache.mina.filter.codec.ProtocolEncoder;
+
+/**
+ * Factory that specifies the encode and decoder to use for parsing XMPP stanzas.
+ *
+ * @author Gaston Dombiak
+ */
+public class XMPPCodecFactory implements ProtocolCodecFactory {
+
+    private final XMPPEncoder encoder;
+    private final XMPPDecoder decoder;
+
+    public XMPPCodecFactory() {
+        encoder = new XMPPEncoder();
+        decoder = new XMPPDecoder();
+    }
+
+    public ProtocolEncoder getEncoder() throws Exception {
+        return encoder;
+    }
+
+    public ProtocolDecoder getDecoder() throws Exception {
+        return decoder;
+    }
+}

src/java/org/jivesoftware/wildfire/nio/XMPPDecoder.java

+/**
+ * $Revision: $
+ * $Date: $
+ *
+ * Copyright (C) 2007 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.nio;
+
+import org.apache.mina.common.ByteBuffer;
+import org.apache.mina.common.IoSession;
+import org.apache.mina.filter.codec.CumulativeProtocolDecoder;
+import org.apache.mina.filter.codec.ProtocolDecoderOutput;
+
+/**
+ * Decoder class that parses ByteBuffers and generates XML stanzas. Generated
+ * stanzas are then passed to the next filters.
+ *
+ * @author Gaston Dombiak
+ */
+public class XMPPDecoder extends CumulativeProtocolDecoder {
+
+    protected boolean doDecode(IoSession session, ByteBuffer in, ProtocolDecoderOutput out)
+            throws Exception {
+        if (in.remaining() < 4) {
+            return false;
+        }
+        // Get the XML light parser from the IoSession
+        XMLLightweightParser parser =
+                (XMLLightweightParser) session.getAttribute(ConnectionHandler.XML_PARSER);
+        // Parse as many stanzas as possible from the received data
+        parser.read(in);
+
+        if (parser.areThereMsgs()) {
+            for (String stanza : parser.getMsgs()) {
+                out.write(stanza);
+            }
+        }
+        return true;
+    }
+}

src/java/org/jivesoftware/wildfire/nio/XMPPEncoder.java

+/**
+ * $Revision: $
+ * $Date: $
+ *
+ * Copyright (C) 2007 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.nio;
+
+import org.apache.mina.common.IoSession;
+import org.apache.mina.filter.codec.ProtocolEncoderAdapter;
+import org.apache.mina.filter.codec.ProtocolEncoderOutput;
+
+/**
+ * Encoder that does nothing. We are already writing ByteBuffers so there is no need
+ * to encode them.<p>
+ *
+ * This class exists as a counterpart of {@link XMPPDecoder}. Unlike that class this class does nothing.
+ *
+ * @author Gaston Dombiak
+ */
+public class XMPPEncoder extends ProtocolEncoderAdapter {
+
+    public void encode(IoSession session, Object message, ProtocolEncoderOutput out)
+            throws Exception {
+        // Ignore. Do nothing. Content being sent is already a bytebuffer (of strings) 
+    }
+}