public void handleAndSendReply(final BufferedInputStream in) 
        throws IOException {

        // Peek at the reply id and tag.
        in.mark(200);
        final XMLInputFactory xif = XMLInputFactory.newInstance();

        final String networkReplyId;
        final boolean question;
        try {
            final XMLStreamReader xmlIn = xif.createXMLStreamReader(in);
            xmlIn.nextTag();
            networkReplyId = xmlIn.getAttributeValue(null, "networkReplyId");
            question = xmlIn.getLocalName().equals("question");
            xmlIn.close();
        } catch (XMLStreamException xme) {
            logger.log(Level.WARNING, "XML stream failure", xme);
            return;
        } 

        // Reset and build a message.
        final DOMMessage msg;
        in.reset();
        try {
            msg = new DOMMessage(in);
        } catch (SAXException e) {
            logger.log(Level.WARNING, "Unable to read message.", e);
            return;
        }

        // Process the message in its own thread.
        final Connection conn = this;
        Thread t = new Thread(msg.getType()) {
                @Override
                public void run() {
                    Element reply, element = msg.getDocument()
                        .getDocumentElement();
                    try {
                        if (question) {
                            reply = messageHandler.handle(conn,
                                (Element)element.getFirstChild());
                            if (reply == null) {
                                reply = DOMMessage.createMessage("reply",
                                    "networkReplyId", networkReplyId);
                            } else {
                                Element header = reply.getOwnerDocument()
                                    .createElement("reply");
                                header.setAttribute("networkReplyId",
                                    networkReplyId);
                                header.appendChild(reply);
                                reply = header;
                            }
                        } else {
                            reply = messageHandler.handle(conn, element);
                        }
                        if (reply != null) conn.sendDumping(reply);
                    } catch (Exception e) {
                        logger.log(Level.WARNING, "Handler failed: "
                            + element.toString(), e);
                    }
                }
            };
        t.setName(name + "-MessageHandler-" + t.getName());
        t.start();
    }