fixed up the test case so its working now along with fixed a schoolboy error in sending messages over XMPP so that works nicely now. FWIW you can now connect via a Jabber client to the broker and interchange messages with the example JMS programs and the web console!

git-svn-id: https://svn.apache.org/repos/asf/incubator/activemq/trunk@468026 13f79535-47bb-0310-9956-ffa450edef68

activemq-xmpp/pom.xml

@@ -177,16 +177,11 @@
         </configuration>
       </plugin>
 
-      <!-- Configure which tests are included/excuded -->
       <plugin>
         <artifactId>maven-surefire-plugin</artifactId>
         <configuration>
-          <!--
-          <includes>
-            <include>**/*Test.*</include>
-          </includes>
-          -->
           <excludes>
+            <!--<exclude>**/XmppTest.*</exclude>-->
           </excludes>
         </configuration>
       </plugin>

activemq-xmpp/src/main/java/org/apache/activemq/transport/xmpp/ProtocolConverter.java

@@ -353,14 +353,21 @@ public class ProtocolConverter {
         item.setNick("broker");
         sendPresence(presence, item);
 
+        /*
         item = new org.jabber.protocol.muc_user.Item();
         item.setAffiliation("admin");
         item.setRole("moderator");
         sendPresence(presence, item);
+        */
 
         // lets create a subscription
         final String to = presence.getTo();
 
+        ActiveMQDestination destination = createActiveMQDestination(to);
+        if (destination == null) {
+            log.debug("No 'to' attribute specified for presence so not creating a JMS subscription");
+            return;
+        }
 
         boolean createConsumer = false;
         ConsumerInfo consumerInfo = null;
@@ -373,6 +380,7 @@ public class ProtocolConverter {
                 ConsumerId consumerId = new ConsumerId(sessionId, consumerIdGenerator.getNextSequenceId());
                 consumerInfo.setConsumerId(consumerId);
                 consumerInfo.setPrefetchSize(10);
+                consumerInfo.setNoLocal(true);
                 createConsumer = true;
             }
         }
@@ -380,7 +388,6 @@ public class ProtocolConverter {
             return;
         }
 
-        ActiveMQDestination destination = createActiveMQDestination(to);
         consumerInfo.setDestination(destination);
 
         subscriptionsByConsumerId.put(consumerInfo.getConsumerId(), new Handler<MessageDispatch>() {
@@ -508,14 +515,16 @@ public class ProtocolConverter {
         activeMQMessage.setTimestamp(System.currentTimeMillis());
         addActiveMQMessageHeaders(activeMQMessage, message);
 
+        /*
         MessageDispatch dispatch = new MessageDispatch();
         dispatch.setDestination(destination);
         dispatch.setMessage(activeMQMessage);
+        */
 
         if (log.isDebugEnabled()) {
             log.debug("Sending ActiveMQ message: " + activeMQMessage);
         }
-        sendToActiveMQ(dispatch, createErrorHandler("send message"));
+        sendToActiveMQ(activeMQMessage, createErrorHandler("send message"));
     }
 
     protected Handler<Response> createErrorHandler(final String text) {

activemq-xmpp/src/main/java/org/apache/activemq/transport/xmpp/XmppTransport.java

@@ -69,7 +69,8 @@ public class XmppTransport extends TcpTransport {
     protected OutputStream outputStream;
     protected InputStream inputStream;
     private ProtocolConverter converter;
-    private String from;
+    private String from = "localhost";
+    private String brokerId = "broker-id-1";
 
     public XmppTransport(WireFormat wireFormat, Socket socket) throws IOException {
         super(wireFormat, socket);
@@ -94,24 +95,10 @@ public class XmppTransport extends TcpTransport {
             if (command instanceof BrokerInfo) {
                 BrokerInfo brokerInfo = (BrokerInfo) command;
 
-                String id = brokerInfo.getBrokerId().toString();
+                brokerId = brokerInfo.getBrokerId().toString();
                 from = brokerInfo.getBrokerName();
                 try {
-                    writeOpenStream(id, from);
+                    writeOpenStream(brokerId, from);
-
-                    // now lets write the features
-                    Features features = new Features();
-
-                    // TODO support TLS
-                    //features.getAny().add(new Starttls());
-
-                    Mechanisms mechanisms = new Mechanisms();
-
-                    // TODO support SASL
-                    //mechanisms.getMechanism().add("DIGEST-MD5");
-                    //mechanisms.getMechanism().add("PLAIN");
-                    features.getAny().add(mechanisms);
-                    marshall(features);
                 }
                 catch (XMLStreamException e) {
                     throw IOExceptionSupport.create(e);
@@ -139,9 +126,14 @@ public class XmppTransport extends TcpTransport {
      * Marshalls the given POJO to the client
      */
     public void marshall(Object command) throws IOException {
+        if (isStopped() || isStopping()) {
+            log.warn("Not marshalling command as shutting down: " + command);
+            return;
+        }
         try {
             marshaller.marshal(command, xmlWriter);
             xmlWriter.flush();
+            outputStream.flush();
         }
         catch (JAXBException e) {
             throw IOExceptionSupport.create(e);
@@ -193,7 +185,8 @@ public class XmppTransport extends TcpTransport {
                     if (event.getEventType() == XMLEvent.END_ELEMENT) {
                         break;
                     }
-                    else if (event.getEventType() == XMLEvent.END_ELEMENT || event.getEventType() == XMLEvent.END_DOCUMENT) {
+                    else
+                    if (event.getEventType() == XMLEvent.END_ELEMENT || event.getEventType() == XMLEvent.END_DOCUMENT) {
                         break;
                     }
                     else {
@@ -250,16 +243,30 @@ public class XmppTransport extends TcpTransport {
     @Override
     protected void initializeStreams() throws Exception {
         // TODO it would be preferable to use class discovery here!
-        context = JAXBContext.newInstance("jabber.client" + ":jabber.server"
+        context = JAXBContext.newInstance("jabber.client"
-                + ":jabber.iq._private" + ":jabber.iq.auth" + ":jabber.iq.gateway" + ":jabber.iq.last" + ":jabber.iq.oob"
+                /*
-                + ":jabber.iq.pass" + ":jabber.iq.roster" + ":jabber.iq.time" + ":jabber.iq.version"
+                + ":jabber.server"
-                + ":org.jabber.etherx.streams" + ":org.jabber.protocol.activity" + ":org.jabber.protocol.address"
+                + ":jabber.iq.gateway"
+                + ":jabber.iq.last"
+                + ":jabber.iq.oob"
+                + ":jabber.iq.pass"
+                + ":jabber.iq.time"
+                + ":jabber.iq.version"
+                + ":org.jabber.protocol.activity" + ":org.jabber.protocol.address"
                 + ":org.jabber.protocol.amp" + ":org.jabber.protocol.amp_errors"
+                + ":org.jabber.protocol.muc_admin"
+                + ":org.jabber.protocol.muc_unique"
+                */
+                + ":jabber.iq._private"
+                + ":jabber.iq.auth"
+                + ":jabber.iq.roster"
+                + ":org.jabber.etherx.streams"
                 + ":org.jabber.protocol.disco_info" + ":org.jabber.protocol.disco_items"
-                + ":org.jabber.protocol.muc" + ":org.jabber.protocol.muc_admin"
+                + ":org.jabber.protocol.muc"
-                + ":org.jabber.protocol.muc_unique" + ":org.jabber.protocol.muc_user"
+                + ":org.jabber.protocol.muc_user"
                 + ":ietf.params.xml.ns.xmpp_sasl" + ":ietf.params.xml.ns.xmpp_stanzas"
-                + ":ietf.params.xml.ns.xmpp_streams" + ":ietf.params.xml.ns.xmpp_tls");
+                + ":ietf.params.xml.ns.xmpp_streams" + ":ietf.params.xml.ns.xmpp_tls"
+        );
 
         inputStream = new TcpBufferedInputStream(socket.getInputStream(), 8 * 1024);
         outputStream = new TcpBufferedOutputStream(socket.getOutputStream(), 16 * 1024);
@@ -270,6 +277,7 @@ public class XmppTransport extends TcpTransport {
     }
 
     protected void writeOpenStream(String id, String from) throws IOException, XMLStreamException {
+        log.debug("Sending initial stream element");
         XMLOutputFactory factory = XMLOutputFactory.newInstance();
         //factory.setProperty(XMLOutputFactory.IS_REPAIRING_NAMESPACES, true);
         xmlWriter = factory.createXMLStreamWriter(outputStream);
@@ -286,7 +294,23 @@ public class XmppTransport extends TcpTransport {
         }
         xmlWriter.writeAttribute("to", to);
         xmlWriter.writeAttribute("from", from);
-        xmlWriter.writeCharacters("\n");
+
+
+        // now lets write the features
+        Features features = new Features();
+
+        // TODO support TLS
+        //features.getAny().add(new Starttls());
+
+        Mechanisms mechanisms = new Mechanisms();
+
+        // TODO support SASL
+        //mechanisms.getMechanism().add("DIGEST-MD5");
+        //mechanisms.getMechanism().add("PLAIN");
+        features.getAny().add(mechanisms);
+        marshall(features);
+
+        log.debug("Initial stream element sent!");
     }
 
 }

activemq-xmpp/src/test/java/org/apache/activemq/transport/xmpp/XmppTest.java

@@ -18,40 +18,25 @@
 package org.apache.activemq.transport.xmpp;
 
 import junit.framework.TestCase;
+import junit.textui.TestRunner;
 import org.jivesoftware.smack.Chat;
-import org.jivesoftware.smack.XMPPException;
 import org.jivesoftware.smack.XMPPConnection;
+import org.jivesoftware.smack.XMPPException;
 
 /**
  * @version $Revision$
  */
 public class XmppTest extends TestCase {
 
+    protected static boolean block = false;
+
     private XmppBroker broker = new XmppBroker();
-    private boolean block = false;
 
     public static void main(String[] args) {
-        XmppTest test = new XmppTest();
+        block = true;
-        test.block = true;
+        TestRunner.run(XmppTest.class);
-        try {
-            test.setUp();
-            test.testConnect();
-        }
-        catch (Exception e) {
-            System.out.println("Caught: " + e);
-            e.printStackTrace();
-        }
-        finally {
-            try {
-                test.tearDown();
-            }
-            catch (Exception e) {
-                System.out.println("Caught: " + e);
-                e.printStackTrace();
-            }
     }
 
-    }
     public void testConnect() throws Exception {
         //ConnectionConfiguration config = new ConnectionConfiguration("localhost", 61222);
         //config.setDebuggerEnabled(true);
@@ -69,9 +54,14 @@ public class XmppTest extends TestCase {
             System.out.println("Sent all messages!");
         }
         catch (XMPPException e) {
+            if (block) {
                 System.out.println("Caught: " + e);
                 e.printStackTrace();
             }
+            else {
+                throw e;
+            }
+        }
         if (block) {
             Thread.sleep(20000);
             System.out.println("Press any key to quit!: ");