activemq-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From bri...@apache.org
Subject svn commit: r431714 - in /incubator/activemq/trunk/activemq-core/src: main/java/org/apache/activemq/transport/stomp/ProtocolConverter.java test/java/org/apache/activemq/transport/stomp/StompTest.java
Date Tue, 15 Aug 2006 21:55:06 GMT
Author: brianm
Date: Tue Aug 15 14:55:06 2006
New Revision: 431714

URL: http://svn.apache.org/viewvc?rev=431714&view=rev
Log:
Stomp should now not pass through headers with null values in the AMQ message (correlation-id,
reply-to, etc)

Modified:
    incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/transport/stomp/ProtocolConverter.java
    incubator/activemq/trunk/activemq-core/src/test/java/org/apache/activemq/transport/stomp/StompTest.java

Modified: incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/transport/stomp/ProtocolConverter.java
URL: http://svn.apache.org/viewvc/incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/transport/stomp/ProtocolConverter.java?rev=431714&r1=431713&r2=431714&view=diff
==============================================================================
--- incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/transport/stomp/ProtocolConverter.java
(original)
+++ incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/transport/stomp/ProtocolConverter.java
Tue Aug 15 14:55:06 2006
@@ -546,15 +546,21 @@
 		
         headers.put(Stomp.Headers.Message.DESTINATION, convertDestination(message.getDestination()));
         headers.put(Stomp.Headers.Message.MESSAGE_ID, message.getJMSMessageID());
-        headers.put(Stomp.Headers.Message.CORRELATION_ID, message.getJMSCorrelationID());
+        if (message.getJMSCorrelationID() != null) {
+            headers.put(Stomp.Headers.Message.CORRELATION_ID, message.getJMSCorrelationID());
+        }
         headers.put(Stomp.Headers.Message.EXPIRATION_TIME, ""+message.getJMSExpiration());
         if (message.getJMSRedelivered()) {
             headers.put(Stomp.Headers.Message.REDELIVERED, "true");
         }
         headers.put(Stomp.Headers.Message.PRORITY, ""+message.getJMSPriority());
-        headers.put(Stomp.Headers.Message.REPLY_TO, convertDestination(message.getJMSReplyTo()));
+        if (message.getJMSReplyTo() != null) {
+            headers.put(Stomp.Headers.Message.REPLY_TO, convertDestination(message.getJMSReplyTo()));
+        }
         headers.put(Stomp.Headers.Message.TIMESTAMP, ""+message.getJMSTimestamp());
-        headers.put(Stomp.Headers.Message.TYPE, message.getJMSType());
+        if (message.getJMSType() != null) {
+            headers.put(Stomp.Headers.Message.TYPE, message.getJMSType());
+        }
 
         // now lets add all the message headers
         Map properties = message.getProperties();

Modified: incubator/activemq/trunk/activemq-core/src/test/java/org/apache/activemq/transport/stomp/StompTest.java
URL: http://svn.apache.org/viewvc/incubator/activemq/trunk/activemq-core/src/test/java/org/apache/activemq/transport/stomp/StompTest.java?rev=431714&r1=431713&r2=431714&view=diff
==============================================================================
--- incubator/activemq/trunk/activemq-core/src/test/java/org/apache/activemq/transport/stomp/StompTest.java
(original)
+++ incubator/activemq/trunk/activemq-core/src/test/java/org/apache/activemq/transport/stomp/StompTest.java
Tue Aug 15 14:55:06 2006
@@ -25,12 +25,7 @@
 import org.apache.activemq.command.ActiveMQTextMessage;
 import org.apache.activemq.transport.stomp.Stomp;
 
-import javax.jms.Connection;
-import javax.jms.JMSException;
-import javax.jms.MessageConsumer;
-import javax.jms.MessageProducer;
-import javax.jms.Session;
-import javax.jms.TextMessage;
+import javax.jms.*;
 
 import java.io.ByteArrayOutputStream;
 import java.io.IOException;
@@ -39,6 +34,8 @@
 import java.net.Socket;
 import java.net.SocketTimeoutException;
 import java.net.URI;
+import java.util.regex.Pattern;
+import java.util.regex.Matcher;
 
 public class StompTest extends CombinationTestSupport {
 
@@ -56,11 +53,11 @@
 
         connector = broker.addConnector("stomp://localhost:0");
         broker.start();
-        
+
         URI connectUri = connector.getConnectUri();
         stompSocket = new Socket("127.0.0.1", connectUri.getPort());
         inputBuffer = new ByteArrayOutputStream();
-        
+
         ActiveMQConnectionFactory cf = new ActiveMQConnectionFactory("vm://localhost");
         connection = cf.createConnection();
         session = connection.createSession(false,Session.AUTO_ACKNOWLEDGE);
@@ -69,7 +66,7 @@
 
 
     }
-    
+
     protected String getQueueName() {
         return getClass().getName() + "." + getName();
     }
@@ -106,7 +103,7 @@
             } else {
                 inputBuffer.write(c);
             }
-        } 
+        }
     }
 
 
@@ -122,43 +119,51 @@
         producer.send(message);
     }
 
+    public void sendBytesMessage(byte[] msg) throws Exception{
+        MessageProducer producer = session.createProducer(queue);
+        BytesMessage message = session.createBytesMessage();
+        message.writeBytes(msg);
+        producer.send(message);
+
+    }
+
     public void testConnect() throws Exception {
-          
+
         String connect_frame = "CONNECT\n" + "login: brianm\n" + "passcode: wombats\n" +
"request-id: 1\n" + "\n" + Stomp.NULL;
         sendFrame(connect_frame);
-     
+
         String f = receiveFrame(10000);
         assertTrue(f.startsWith("CONNECTED"));
         assertTrue(f.indexOf("response-id:1") >= 0);
-        
+
     }
-    
+
     public void testSendMessage() throws Exception {
-        
+
         MessageConsumer consumer = session.createConsumer(queue);
-        
-        String frame = 
-            "CONNECT\n" + 
-            "login: brianm\n" + 
+
+        String frame =
+            "CONNECT\n" +
+            "login: brianm\n" +
             "passcode: wombats\n\n"+
             Stomp.NULL;
         sendFrame(frame);
-        
+
         frame = receiveFrame(10000);
         assertTrue(frame.startsWith("CONNECTED"));
-        
+
         frame =
             "SEND\n" +
             "destination:/queue/" + getQueueName() + "\n\n" +
             "Hello World" +
             Stomp.NULL;
-        
+
         sendFrame(frame);
-        
+
         TextMessage message = (TextMessage) consumer.receive(1000);
         assertNotNull(message);
         assertEquals("Hello World", message.getText());
-        
+
         // Make sure that the timestamp is valid - should
         // be very close to the current time.
         long tnow = System.currentTimeMillis();
@@ -166,47 +171,46 @@
         assertTrue( Math.abs(tnow - tmsg) < 1000 );
     }
 
-    
     public void testJMSXGroupIdCanBeSet() throws Exception {
-        
+
         MessageConsumer consumer = session.createConsumer(queue);
-        
-        String frame = 
-            "CONNECT\n" + 
-            "login: brianm\n" + 
+
+        String frame =
+            "CONNECT\n" +
+            "login: brianm\n" +
             "passcode: wombats\n\n"+
             Stomp.NULL;
         sendFrame(frame);
-        
+
         frame = receiveFrame(10000);
         assertTrue(frame.startsWith("CONNECTED"));
-        
+
         frame =
             "SEND\n" +
             "destination:/queue/" + getQueueName() + "\n" +
             "JMSXGroupID: TEST\n\n" +
             "Hello World" +
             Stomp.NULL;
-        
+
         sendFrame(frame);
-        
+
         TextMessage message = (TextMessage) consumer.receive(1000);
         assertNotNull(message);
         assertEquals("TEST", ((ActiveMQTextMessage)message).getGroupID());
     }
 
-    
+
     public void testSendMessageWithCustomHeadersAndSelector() throws Exception {
-        
+
         MessageConsumer consumer = session.createConsumer(queue, "foo = 'abc'");
-        
-        String frame = 
-            "CONNECT\n" + 
-            "login: brianm\n" + 
+
+        String frame =
+            "CONNECT\n" +
+            "login: brianm\n" +
             "passcode: wombats\n\n"+
             Stomp.NULL;
         sendFrame(frame);
-     
+
         frame = receiveFrame(10000);
         assertTrue(frame.startsWith("CONNECTED"));
 
@@ -226,18 +230,18 @@
         assertEquals("foo", "abc", message.getStringProperty("foo"));
         assertEquals("bar", "123", message.getStringProperty("bar"));
     }
-    
+
     public void testSendMessageWithStandardHeaders() throws Exception {
-        
+
         MessageConsumer consumer = session.createConsumer(queue);
-        
-        String frame = 
-            "CONNECT\n" + 
-            "login: brianm\n" + 
+
+        String frame =
+            "CONNECT\n" +
+            "login: brianm\n" +
             "passcode: wombats\n\n"+
             Stomp.NULL;
         sendFrame(frame);
-     
+
         frame = receiveFrame(10000);
         assertTrue(frame.startsWith("CONNECTED"));
 
@@ -263,36 +267,75 @@
         assertEquals("getJMSPriority", 3, message.getJMSPriority());
         assertEquals("foo", "abc", message.getStringProperty("foo"));
         assertEquals("bar", "123", message.getStringProperty("bar"));
-        
+
         assertEquals("JMSXGroupID", "abc", message.getStringProperty("JMSXGroupID"));
         ActiveMQTextMessage amqMessage = (ActiveMQTextMessage) message;
         assertEquals("GroupID", "abc", amqMessage.getGroupID());
     }
-    
+
     public void testSubscribeWithAutoAck() throws Exception {
-        
+
         String frame =
             "CONNECT\n" +
             "login: brianm\n" +
             "passcode: wombats\n\n"+
             Stomp.NULL;
         sendFrame(frame);
-        
+
         frame = receiveFrame(100000);
         assertTrue(frame.startsWith("CONNECTED"));
-        
+
         frame =
             "SUBSCRIBE\n" +
             "destination:/queue/" + getQueueName() + "\n" +
             "ack:auto\n\n" +
             Stomp.NULL;
         sendFrame(frame);
-        
+
         sendMessage(getName());
-        
+
+        frame = receiveFrame(10000);
+        assertTrue(frame.startsWith("MESSAGE"));
+
+        frame =
+            "DISCONNECT\n" +
+            "\n\n"+
+            Stomp.NULL;
+        sendFrame(frame);
+    }
+
+        public void testSubscribeWithAutoAckAndBytesMessage() throws Exception {
+
+        String frame =
+            "CONNECT\n" +
+            "login: brianm\n" +
+            "passcode: wombats\n\n"+
+            Stomp.NULL;
+        sendFrame(frame);
+
+        frame = receiveFrame(100000);
+        assertTrue(frame.startsWith("CONNECTED"));
+
+        frame =
+            "SUBSCRIBE\n" +
+            "destination:/queue/" + getQueueName() + "\n" +
+            "ack:auto\n\n" +
+            Stomp.NULL;
+        sendFrame(frame);
+
+        sendBytesMessage(new byte[] {1,2,3,4,5});
+
         frame = receiveFrame(10000);
         assertTrue(frame.startsWith("MESSAGE"));
-        
+
+        Pattern cl = Pattern.compile("Content-length:\\s*(\\d+)", Pattern.CASE_INSENSITIVE);
+        Matcher cl_matcher = cl.matcher(frame);
+        assertTrue(cl_matcher.find());
+        assertEquals("5", cl_matcher.group(1));
+
+        Pattern null_type_pattern = Pattern.compile("type:\\s*null", Pattern.CASE_INSENSITIVE);
+        assertFalse(null_type_pattern.matcher(frame).find());
+
         frame =
             "DISCONNECT\n" +
             "\n\n"+
@@ -301,25 +344,25 @@
     }
 
     public void testSubscribeWithMessageSentWithProperties() throws Exception {
-        
+
         String frame =
             "CONNECT\n" +
             "login: brianm\n" +
             "passcode: wombats\n\n"+
             Stomp.NULL;
         sendFrame(frame);
-        
+
         frame = receiveFrame(100000);
         assertTrue(frame.startsWith("CONNECTED"));
-        
+
         frame =
             "SUBSCRIBE\n" +
             "destination:/queue/" + getQueueName() + "\n" +
             "ack:auto\n\n" +
             Stomp.NULL;
         sendFrame(frame);
-        
-        
+
+
         MessageProducer producer = session.createProducer(queue);
         TextMessage message = session.createTextMessage("Hello World");
         message.setStringProperty("s", "value");
@@ -331,12 +374,12 @@
         message.setLongProperty("l", 121);
         message.setShortProperty("s", (short) 12);
         producer.send(message);
-        
+
         frame = receiveFrame(10000);
         assertTrue(frame.startsWith("MESSAGE"));
-        
+
         System.out.println("out: "+frame);
-        
+
         frame =
             "DISCONNECT\n" +
             "\n\n"+
@@ -507,7 +550,7 @@
         sendFrame(frame);
 
         waitForFrameToTakeEffect();
-        
+
         //send a message to our queue
         sendMessage("second message");
 
@@ -535,7 +578,7 @@
 
         String f = receiveFrame(1000);
         assertTrue(f.startsWith("CONNECTED"));
-        
+
         frame =
             "BEGIN\n" +
             "transaction: tx1\n" +
@@ -560,7 +603,7 @@
         sendFrame(frame);
 
         waitForFrameToTakeEffect();
-        
+
         TextMessage message = (TextMessage) consumer.receive(1000);
         assertNotNull("Should have received a message", message);
     }



Mime
View raw message