6:09 p.m.
Author: gaohoward
Date: 2011-09-23 12:09:48 -0400 (Fri, 23 Sep 2011)
New Revision: 11404
Modified:
branches/STOMP11/hornetq-core/src/main/java/org/hornetq/core/protocol/stomp/v10/StompFrameHandlerV10.java
branches/STOMP11/tests/integration-tests/src/test/java/org/hornetq/tests/integration/stomp/util/AbstractClientStompFrame.java
branches/STOMP11/tests/integration-tests/src/test/java/org/hornetq/tests/integration/stomp/util/AbstractStompClientConnection.java
branches/STOMP11/tests/integration-tests/src/test/java/org/hornetq/tests/integration/stomp/util/ClientStompFrame.java
branches/STOMP11/tests/integration-tests/src/test/java/org/hornetq/tests/integration/stomp/util/StompClientConnection.java
branches/STOMP11/tests/integration-tests/src/test/java/org/hornetq/tests/integration/stomp/v11/StompTestV11.java
Log:
tests
Modified:
branches/STOMP11/hornetq-core/src/main/java/org/hornetq/core/protocol/stomp/v10/StompFrameHandlerV10.java
===================================================================
---
branches/STOMP11/hornetq-core/src/main/java/org/hornetq/core/protocol/stomp/v10/StompFrameHandlerV10.java 2011-09-23
13:53:53 UTC (rev 11403)
+++
branches/STOMP11/hornetq-core/src/main/java/org/hornetq/core/protocol/stomp/v10/StompFrameHandlerV10.java 2011-09-23
16:09:48 UTC (rev 11404)
@@ -13,7 +13,6 @@
package org.hornetq.core.protocol.stomp.v10;
import java.io.UnsupportedEncodingException;
-import java.util.List;
import java.util.Map;
import org.hornetq.api.core.HornetQBuffer;
@@ -27,7 +26,6 @@
import org.hornetq.core.protocol.stomp.StompConnection;
import org.hornetq.core.protocol.stomp.StompDecoder;
import org.hornetq.core.protocol.stomp.StompFrame;
-import org.hornetq.core.protocol.stomp.StompFrame.Header;
import org.hornetq.core.protocol.stomp.StompSubscription;
import org.hornetq.core.protocol.stomp.StompUtils;
import org.hornetq.core.protocol.stomp.VersionedStompFrameHandler;
@@ -53,7 +51,6 @@
@Override
public StompFrame onConnect(StompFrame frame)
{
- log.error("-----------------onConnection ()");
StompFrame response = null;
Map<String, String> headers = frame.getHeadersMap();
String login = (String)headers.get(Stomp.Headers.Connect.LOGIN);
@@ -61,10 +58,8 @@
String clientID = (String)headers.get(Stomp.Headers.Connect.CLIENT_ID);
String requestID = (String)headers.get(Stomp.Headers.Connect.REQUEST_ID);
- log.error("------------ validating user: " + login + " code " +
passcode);
if (connection.validateUser(login, passcode))
{
- log.error("-------user OK!!!");
connection.setClientID(clientID);
connection.setValid(true);
@@ -84,7 +79,6 @@
}
else
{
- log.error("--------user NOT ok!!");
//not valid
response = new StompFrameV10(Stomp.Responses.ERROR);
response.addHeader(Stomp.Headers.Error.MESSAGE, "Failed to connect");
@@ -95,7 +89,6 @@
catch (UnsupportedEncodingException e)
{
log.error("Encoding problem", e);
- //then we will send a null body message.
}
}
return response;
Modified:
branches/STOMP11/tests/integration-tests/src/test/java/org/hornetq/tests/integration/stomp/util/AbstractClientStompFrame.java
===================================================================
---
branches/STOMP11/tests/integration-tests/src/test/java/org/hornetq/tests/integration/stomp/util/AbstractClientStompFrame.java 2011-09-23
13:53:53 UTC (rev 11403)
+++
branches/STOMP11/tests/integration-tests/src/test/java/org/hornetq/tests/integration/stomp/util/AbstractClientStompFrame.java 2011-09-23
16:09:48 UTC (rev 11404)
@@ -70,7 +70,7 @@
}
sb.append((char)0);
- String data = new String(sb.toString());
+ String data = sb.toString();
byte[] byteValue = data.getBytes("UTF-8");
@@ -82,6 +82,35 @@
}
@Override
+ public ByteBuffer toByteBufferWithExtra(String str) throws
UnsupportedEncodingException
+ {
+ StringBuffer sb = new StringBuffer();
+ sb.append(command + "\n");
+ int n = headers.size();
+ for (int i = 0; i < n; i++)
+ {
+ sb.append(headers.get(i).key + ":" + headers.get(i).val +
"\n");
+ }
+ sb.append("\n");
+ if (body != null)
+ {
+ sb.append(body);
+ }
+ sb.append((char)0);
+ sb.append(str);
+
+ String data = sb.toString();
+
+ byte[] byteValue = data.getBytes("UTF-8");
+
+ ByteBuffer buffer = ByteBuffer.allocateDirect(byteValue.length);
+ buffer.put(byteValue);
+
+ buffer.rewind();
+ return buffer;
+ }
+
+ @Override
public boolean needsReply()
{
if ("CONNECT".equals(command) || headerKeys.contains(HEADER_RECEIPT))
Modified:
branches/STOMP11/tests/integration-tests/src/test/java/org/hornetq/tests/integration/stomp/util/AbstractStompClientConnection.java
===================================================================
---
branches/STOMP11/tests/integration-tests/src/test/java/org/hornetq/tests/integration/stomp/util/AbstractStompClientConnection.java 2011-09-23
13:53:53 UTC (rev 11403)
+++
branches/STOMP11/tests/integration-tests/src/test/java/org/hornetq/tests/integration/stomp/util/AbstractStompClientConnection.java 2011-09-23
16:09:48 UTC (rev 11404)
@@ -113,7 +113,38 @@
}
return response;
}
-
+
+ public ClientStompFrame sendWickedFrame(ClientStompFrame frame) throws IOException,
InterruptedException
+ {
+ ClientStompFrame response = null;
+ ByteBuffer buffer = frame.toByteBufferWithExtra("\n");
+
+ while (buffer.remaining() > 0)
+ {
+ socketChannel.write(buffer);
+ }
+
+ //now response
+ if (frame.needsReply())
+ {
+ response = receiveFrame();
+
+ //filter out server ping
+ while (response != null)
+ {
+ if (response.getCommand().equals("STOMP"))
+ {
+ response = receiveFrame();
+ }
+ else
+ {
+ break;
+ }
+ }
+ }
+ return response;
+ }
+
public ClientStompFrame receiveFrame() throws InterruptedException
{
return frameQueue.poll(10, TimeUnit.SECONDS);
@@ -138,9 +169,16 @@
frameBytes[j] = receiveList.get(j);
}
ClientStompFrame frame = factory.createFrame(new String(frameBytes,
"UTF-8"));
- frameQueue.offer(frame);
- receiveList.clear();
+ if (validateFrame(frame))
+ {
+ frameQueue.offer(frame);
+ receiveList.clear();
+ }
+ else
+ {
+ receiveList.add(b);
+ }
}
}
else
@@ -152,6 +190,20 @@
readBuffer.rewind();
}
+ private boolean validateFrame(ClientStompFrame f) throws UnsupportedEncodingException
+ {
+ String h = f.getHeader("content-length");
+ if (h != null)
+ {
+ int len = Integer.valueOf(h);
+ if (f.getBody().getBytes("UTF-8").length < len)
+ {
+ return false;
+ }
+ }
+ return true;
+ }
+
protected void close() throws IOException
{
socketChannel.close();
Modified:
branches/STOMP11/tests/integration-tests/src/test/java/org/hornetq/tests/integration/stomp/util/ClientStompFrame.java
===================================================================
---
branches/STOMP11/tests/integration-tests/src/test/java/org/hornetq/tests/integration/stomp/util/ClientStompFrame.java 2011-09-23
13:53:53 UTC (rev 11403)
+++
branches/STOMP11/tests/integration-tests/src/test/java/org/hornetq/tests/integration/stomp/util/ClientStompFrame.java 2011-09-23
16:09:48 UTC (rev 11404)
@@ -39,5 +39,7 @@
public String getHeader(String header);
public String getBody();
+
+ public ByteBuffer toByteBufferWithExtra(String str) throws
UnsupportedEncodingException;
}
Modified:
branches/STOMP11/tests/integration-tests/src/test/java/org/hornetq/tests/integration/stomp/util/StompClientConnection.java
===================================================================
---
branches/STOMP11/tests/integration-tests/src/test/java/org/hornetq/tests/integration/stomp/util/StompClientConnection.java 2011-09-23
13:53:53 UTC (rev 11403)
+++
branches/STOMP11/tests/integration-tests/src/test/java/org/hornetq/tests/integration/stomp/util/StompClientConnection.java 2011-09-23
16:09:48 UTC (rev 11404)
@@ -48,6 +48,8 @@
void stopPinger();
void destroy();
+
+ ClientStompFrame sendWickedFrame(ClientStompFrame frame) throws IOException,
InterruptedException;
}
Modified:
branches/STOMP11/tests/integration-tests/src/test/java/org/hornetq/tests/integration/stomp/v11/StompTestV11.java
===================================================================
---
branches/STOMP11/tests/integration-tests/src/test/java/org/hornetq/tests/integration/stomp/v11/StompTestV11.java 2011-09-23
13:53:53 UTC (rev 11403)
+++
branches/STOMP11/tests/integration-tests/src/test/java/org/hornetq/tests/integration/stomp/v11/StompTestV11.java 2011-09-23
16:09:48 UTC (rev 11404)
@@ -22,6 +22,7 @@
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
+import javax.jms.BytesMessage;
import javax.jms.Message;
import javax.jms.MessageConsumer;
import javax.jms.MessageListener;
@@ -1210,7 +1211,104 @@
connV11.disconnect();
}
+ public void testSendMessage() throws Exception
+ {
+ MessageConsumer consumer = session.createConsumer(queue);
+ connV11.connect(defUser, defPass);
+
+ ClientStompFrame frame = connV11.createFrame("SEND");
+ frame.addHeader("destination", getQueuePrefix() + getQueueName());
+ frame.setBody("Hello World");
+
+ connV11.sendFrame(frame);
+
+ TextMessage message = (TextMessage)consumer.receive(1000);
+ Assert.assertNotNull(message);
+ Assert.assertEquals("Hello World", message.getText());
+ // Assert default priority 4 is used when priority header is not set
+ Assert.assertEquals("getJMSPriority", 4, message.getJMSPriority());
+
+ // Make sure that the timestamp is valid - should
+ // be very close to the current time.
+ long tnow = System.currentTimeMillis();
+ long tmsg = message.getJMSTimestamp();
+ Assert.assertTrue(Math.abs(tnow - tmsg) < 1000);
+ }
+
+ public void testSendMessageWithContentLength() throws Exception
+ {
+ MessageConsumer consumer = session.createConsumer(queue);
+
+ connV11.connect(defUser, defPass);
+
+ byte[] data = new byte[] { 1, 0, 0, 4 };
+
+ ClientStompFrame frame = connV11.createFrame("SEND");
+
+ frame.addHeader("destination", getQueuePrefix() + getQueueName());
+ frame.setBody(new String(data, "UTF-8"));
+
+ frame.addHeader("content-length", String.valueOf(data.length));
+
+ connV11.sendFrame(frame);
+
+ BytesMessage message = (BytesMessage)consumer.receive(10000);
+ Assert.assertNotNull(message);
+ //there is one extra null byte
+ assertEquals(data.length + 1, message.getBodyLength());
+ assertEquals(data[0], message.readByte());
+ assertEquals(data[1], message.readByte());
+ assertEquals(data[2], message.readByte());
+ assertEquals(data[3], message.readByte());
+ }
+
+ public void testSendMessageWithCustomHeadersAndSelector() throws Exception
+ {
+ MessageConsumer consumer = session.createConsumer(queue, "foo =
'abc'");
+
+ connV11.connect(defUser, defPass);
+
+ ClientStompFrame frame = connV11.createFrame("SEND");
+ frame.addHeader("foo", "abc");
+ frame.addHeader("bar", "123");
+
+ frame.addHeader("destination", getQueuePrefix() + getQueueName());
+ frame.setBody("Hello World");
+
+ connV11.sendFrame(frame);
+
+ TextMessage message = (TextMessage)consumer.receive(1000);
+ Assert.assertNotNull(message);
+ Assert.assertEquals("Hello World", message.getText());
+ Assert.assertEquals("foo", "abc",
message.getStringProperty("foo"));
+ Assert.assertEquals("bar", "123",
message.getStringProperty("bar"));
+ }
+
+ public void testSendMessageWithLeadingNewLine() throws Exception
+ {
+ MessageConsumer consumer = session.createConsumer(queue);
+
+ connV11.connect(defUser, defPass);
+
+ ClientStompFrame frame = connV11.createFrame("SEND");
+
+ frame.addHeader("destination", getQueuePrefix() + getQueueName());
+ frame.setBody("Hello World");
+
+ connV11.sendWickedFrame(frame);
+
+ TextMessage message = (TextMessage)consumer.receive(1000);
+ Assert.assertNotNull(message);
+ Assert.assertEquals("Hello World", message.getText());
+
+ // Make sure that the timestamp is valid - should
+ // be very close to the current time.
+ long tnow = System.currentTimeMillis();
+ long tmsg = message.getJMSTimestamp();
+ Assert.assertTrue(Math.abs(tnow - tmsg) < 1000);
+ }
+
//-----------------private help methods
private void abortTransaction(StompClientConnection conn, String txID) throws
IOException, InterruptedException