12:06 a.m.
Author: gaohoward
Date: 2012-02-21 01:06:47 -0500 (Tue, 21 Feb 2012)
New Revision: 12153
Modified:
trunk/tests/concurrent-tests/src/test/java/org/hornetq/tests/concurrent/stomp/ConcurrentStompTest.java
Log:
fix stomp test failure
Modified:
trunk/tests/concurrent-tests/src/test/java/org/hornetq/tests/concurrent/stomp/ConcurrentStompTest.java
===================================================================
---
trunk/tests/concurrent-tests/src/test/java/org/hornetq/tests/concurrent/stomp/ConcurrentStompTest.java 2012-02-21
01:42:02 UTC (rev 12152)
+++
trunk/tests/concurrent-tests/src/test/java/org/hornetq/tests/concurrent/stomp/ConcurrentStompTest.java 2012-02-21
06:06:47 UTC (rev 12153)
@@ -34,33 +34,31 @@
import org.hornetq.core.server.HornetQServer;
import org.hornetq.core.server.HornetQServers;
import org.hornetq.spi.core.protocol.ProtocolType;
+import org.hornetq.tests.integration.stomp.StompTestBase;
import org.hornetq.tests.util.UnitTestCase;
-public class ConcurrentStompTest extends UnitTestCase
+public class ConcurrentStompTest extends StompTestBase
{
- private final int port = 61613;
-
- private Socket stompSocket;
-
- private ByteArrayOutputStream inputBuffer;
-
private Socket stompSocket_2;
private ByteArrayOutputStream inputBuffer_2;
- private HornetQServer server;
-
/**
* Send messages on 1 socket and receives them concurrently on another socket.
*/
public void testSendManyMessages() throws Exception
{
+ try
+ {
String connect = "CONNECT\n" + "login: brianm\n" +
"passcode: wombats\n\n" + Stomp.NULL;
- sendFrame(stompSocket, connect);
- String connected = receiveFrame(stompSocket, inputBuffer, 10000);
+ sendFrame(connect);
+ String connected = receiveFrame(10000);
Assert.assertTrue(connected.startsWith("CONNECTED"));
+ stompSocket_2 = createSocket();
+ inputBuffer_2 = new ByteArrayOutputStream();
+
sendFrame(stompSocket_2, connect);
connected = receiveFrame(stompSocket_2, inputBuffer_2, 10000);
Assert.assertTrue(connected.startsWith("CONNECTED"));
@@ -70,7 +68,7 @@
String subscribe =
"SUBSCRIBE\n" +
- "destination:" + getQueueName() + "\n" +
+ "destination:" + getQueuePrefix() + getQueueName() + "\n" +
"ack:auto\n\n" +
Stomp.NULL;
sendFrame(stompSocket_2, subscribe);
@@ -100,89 +98,41 @@
};
}.start();
- String send = "SEND\n" + "destination:" + getQueueName() +
"\n";
+ String send = "SEND\n" + "destination:" + getQueuePrefix() +
getQueueName() + "\n";
for (int i = 1; i <= count; i++)
{
// Thread.sleep(1);
System.out.println(">>> " + i);
- sendFrame(stompSocket, send + "count:" + i + "\n\n" +
Stomp.NULL);
+ sendFrame(send + "count:" + i + "\n\n" + Stomp.NULL);
}
assertTrue(latch.await(60, TimeUnit.SECONDS));
-
- }
-
- // Implementation methods
- // -------------------------------------------------------------------------
- @Override
- protected void setUp() throws Exception
- {
- super.setUp();
-
- server = createServer();
- server.start();
-
- stompSocket = createSocket();
- inputBuffer = new ByteArrayOutputStream();
- stompSocket_2 = createSocket();
- inputBuffer_2 = new ByteArrayOutputStream();
-
- }
-
- private HornetQServer createServer() throws Exception
- {
- Configuration config = createBasicConfig();
- config.setSecurityEnabled(false);
- config.setPersistenceEnabled(false);
-
- Map<String, Object> params = new HashMap<String, Object>();
- params.put(TransportConstants.PROTOCOL_PROP_NAME, ProtocolType.STOMP.toString());
- params.put(TransportConstants.PORT_PROP_NAME,
TransportConstants.DEFAULT_STOMP_PORT);
- TransportConfiguration stompTransport = new
TransportConfiguration(NettyAcceptorFactory.class.getName(), params);
- config.getAcceptorConfigurations().add(stompTransport);
- config.getAcceptorConfigurations().add(new
TransportConfiguration(InVMAcceptorFactory.class.getName()));
- config.getQueueConfigurations().add(new CoreQueueConfiguration(getQueueName(),
getQueueName(), null, false));
- return addServer(HornetQServers.newHornetQServer(config));
- }
-
- @Override
- protected void tearDown() throws Exception
- {
- if (stompSocket != null)
- {
- stompSocket.close();
+
}
-
- if (stompSocket_2 != null)
+ finally
{
stompSocket_2.close();
+ inputBuffer_2.close();
}
+
+
- super.tearDown();
}
- protected Socket createSocket() throws IOException
- {
- return new Socket("127.0.0.1", port);
- }
-
- protected String getQueueName()
- {
- return "test";
- }
-
+ // Implementation methods
+ // -------------------------------------------------------------------------
public void sendFrame(Socket socket, String data) throws Exception
{
byte[] bytes = data.getBytes("UTF-8");
OutputStream outputStream = socket.getOutputStream();
- for (byte b : bytes)
+ for (int i = 0; i < bytes.length; i++)
{
- outputStream.write(b);
+ outputStream.write(bytes[i]);
}
outputStream.flush();
}
-
- public String receiveFrame(Socket socket, ByteArrayOutputStream inputBuffer, long
timeOut) throws Exception
+
+ public String receiveFrame(Socket socket, ByteArrayOutputStream input, long timeOut)
throws Exception
{
socket.setSoTimeout((int)timeOut);
InputStream is = socket.getInputStream();
@@ -199,18 +149,19 @@
c = is.read();
if (c != '\n')
{
- byte[] ba = inputBuffer.toByteArray();
+ byte[] ba = input.toByteArray();
System.out.println(new String(ba, "UTF-8"));
}
Assert.assertEquals("Expecting stomp frame to terminate with \0\n",
c, '\n');
- byte[] ba = inputBuffer.toByteArray();
- inputBuffer.reset();
+ byte[] ba = input.toByteArray();
+ input.reset();
return new String(ba, "UTF-8");
}
else
{
- inputBuffer.write(c);
+ input.write(c);
}
}
}
+
}