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);
}
