6:11 p.m.
Author: jmesnil
Date: 2010-01-20 12:11:18 -0500 (Wed, 20 Jan 2010)
New Revision: 8817
Modified:
branches/HORNETQ-129_STOMP_protocol/src/main/org/hornetq/integration/stomp/Stomp.java
branches/HORNETQ-129_STOMP_protocol/src/main/org/hornetq/integration/transports/netty/NettyAcceptor.java
branches/HORNETQ-129_STOMP_protocol/tests/src/org/hornetq/tests/integration/stomp/StompTest.java
Log:
https://jira.jboss.org/jira/browse/HORNETQ-129: Implement STOMP v1.0
* added SUBSCRIBE command
Modified:
branches/HORNETQ-129_STOMP_protocol/src/main/org/hornetq/integration/stomp/Stomp.java
===================================================================
---
branches/HORNETQ-129_STOMP_protocol/src/main/org/hornetq/integration/stomp/Stomp.java 2010-01-20
17:07:39 UTC (rev 8816)
+++
branches/HORNETQ-129_STOMP_protocol/src/main/org/hornetq/integration/stomp/Stomp.java 2010-01-20
17:11:18 UTC (rev 8817)
@@ -31,8 +31,8 @@
String CONNECT = "CONNECT";
String SEND = "SEND";
String DISCONNECT = "DISCONNECT";
- String SUBSCRIBE = "SUB";
- String UNSUBSCRIBE = "UNSUB";
+ String SUBSCRIBE = "SUBSCRIBE";
+ String UNSUBSCRIBE = "UNSUBSCRIBE";
String BEGIN_TRANSACTION = "BEGIN";
String COMMIT_TRANSACTION = "COMMIT";
String ABORT_TRANSACTION = "ABORT";
Modified:
branches/HORNETQ-129_STOMP_protocol/src/main/org/hornetq/integration/transports/netty/NettyAcceptor.java
===================================================================
---
branches/HORNETQ-129_STOMP_protocol/src/main/org/hornetq/integration/transports/netty/NettyAcceptor.java 2010-01-20
17:07:39 UTC (rev 8816)
+++
branches/HORNETQ-129_STOMP_protocol/src/main/org/hornetq/integration/transports/netty/NettyAcceptor.java 2010-01-20
17:11:18 UTC (rev 8817)
@@ -43,6 +43,9 @@
import org.hornetq.core.remoting.RemotingConnection;
import org.hornetq.core.remoting.impl.CorePacketDecoder;
import org.hornetq.core.remoting.impl.ssl.SSLSupport;
+import org.hornetq.core.remoting.impl.wireformat.PacketImpl;
+import org.hornetq.core.remoting.impl.wireformat.SessionConsumerFlowCreditMessage;
+import org.hornetq.core.remoting.impl.wireformat.SessionCreateConsumerMessage;
import org.hornetq.core.remoting.impl.wireformat.SessionSendMessage;
import org.hornetq.core.server.HornetQServer;
import org.hornetq.core.server.ServerMessage;
@@ -587,7 +590,7 @@
{
response = onConnect(frame, server, connection);
}
- if (Stomp.Commands.DISCONNECT.equals(command))
+ else if (Stomp.Commands.DISCONNECT.equals(command))
{
response = onDisconnect(frame, server, connection);
}
@@ -595,10 +598,16 @@
{
response = onSend(frame, server, connection);
}
+ else if (Stomp.Commands.SUBSCRIBE.equals(command))
+ {
+ response = onSubscribe(frame, server, connection);
+ }
else
{
log.error("Unsupported Stomp frame: " + frame);
+ response = new StompFrame(Stomp.Responses.ERROR, new HashMap<String,
Object>(), ("Unsupported frame: " + command).getBytes());
}
+
if (response != null)
{
System.out.println(">>> will reply " + response);
@@ -638,19 +647,44 @@
}
}
- private void checkConnected(RemotingConnection connection) throws StompException
+ /**
+ * @param frame
+ * @param server
+ * @param connection
+ * @return
+ * @throws StompException
+ * @throws HornetQException
+ */
+ private StompFrame onSubscribe(StompFrame frame, HornetQServer server,
RemotingConnection connection) throws StompException, HornetQException
{
+ Map<String, Object> headers = frame.getHeaders();
+ String queue = (String)headers.get(Stomp.Headers.Send.DESTINATION);
+ SimpleString queueName = StompDestinationConverter.convertDestination(queue);
+
+ ServerSession session = checkAndGetSession(connection);
+ long id = server.getStorageManager().generateUniqueID();
+ SessionCreateConsumerMessage packet = new SessionCreateConsumerMessage(id ,
queueName, null, false, false);
+ session.handleCreateConsumer(packet);
+ SessionConsumerFlowCreditMessage credits = new
SessionConsumerFlowCreditMessage(id, -1);
+ session.handleReceiveConsumerCredits(credits );
+ session.handleStart(new PacketImpl(PacketImpl.SESS_START));
+
+ return null;
+ }
+
+ private ServerSession checkAndGetSession(RemotingConnection connection) throws
StompException
+ {
ServerSession session = sessions.get(connection);
if (session == null)
{
throw new StompException("Not connected");
}
+ return session;
}
+
private StompFrame onDisconnect(StompFrame frame, HornetQServer server,
RemotingConnection connection) throws StompException
{
- checkConnected(connection);
-
- ServerSession session = sessions.get(connection);
+ ServerSession session = checkAndGetSession(connection);
if (session != null)
{
try
@@ -668,7 +702,7 @@
private StompFrame onSend(StompFrame frame, HornetQServer server,
RemotingConnection connection) throws HornetQException, StompException
{
- checkConnected(connection);
+ ServerSession session = checkAndGetSession(connection);
Map<String, Object> headers = frame.getHeaders();
String queue = (String)headers.get(Stomp.Headers.Send.DESTINATION);
@@ -703,7 +737,6 @@
message.getBodyBuffer().writeBytes(content);
}
- ServerSession session = sessions.get(connection);
SessionSendMessage packet = new SessionSendMessage(message, false);
session.handleSend(packet);
if (headers.containsKey(Stomp.Headers.RECEIPT_REQUESTED))
Modified:
branches/HORNETQ-129_STOMP_protocol/tests/src/org/hornetq/tests/integration/stomp/StompTest.java
===================================================================
---
branches/HORNETQ-129_STOMP_protocol/tests/src/org/hornetq/tests/integration/stomp/StompTest.java 2010-01-20
17:07:39 UTC (rev 8816)
+++
branches/HORNETQ-129_STOMP_protocol/tests/src/org/hornetq/tests/integration/stomp/StompTest.java 2010-01-20
17:11:18 UTC (rev 8817)
@@ -728,6 +728,10 @@
Stomp.NULL;
sendFrame(frame);
+ waitForFrameToTakeEffect();
+ // check the message is not committed
+ assertNull(consumer.receive(100));
+
frame =
"COMMIT\n" +
"transaction: tx1\n" +
@@ -899,7 +903,6 @@
int c = 0;
for (; ;) {
c = is.read();
- System.out.println(c);
if (c < 0) {
throw new IOException("socket closed.");
}