[hornetq-commits] JBoss hornetq SVN: r11363 - in branches/STOMP11: hornetq-core/src/main/java/org/hornetq/core/protocol/core/impl and 5 other directories.
do-not-reply at jboss.org
do-not-reply at jboss.org
Mon Sep 19 09:17:33 EDT 2011
Author: gaohoward
Date: 2011-09-19 09:17:33 -0400 (Mon, 19 Sep 2011)
New Revision: 11363
Modified:
branches/STOMP11/hornetq-core/src/main/java/org/hornetq/core/postoffice/impl/BindingsImpl.java
branches/STOMP11/hornetq-core/src/main/java/org/hornetq/core/postoffice/impl/PostOfficeImpl.java
branches/STOMP11/hornetq-core/src/main/java/org/hornetq/core/protocol/core/impl/ChannelImpl.java
branches/STOMP11/hornetq-core/src/main/java/org/hornetq/core/protocol/core/impl/CoreSessionCallback.java
branches/STOMP11/hornetq-core/src/main/java/org/hornetq/core/protocol/stomp/StompConnection.java
branches/STOMP11/hornetq-core/src/main/java/org/hornetq/core/protocol/stomp/StompDecoder.java
branches/STOMP11/hornetq-core/src/main/java/org/hornetq/core/protocol/stomp/StompFrame.java
branches/STOMP11/hornetq-core/src/main/java/org/hornetq/core/protocol/stomp/StompSession.java
branches/STOMP11/hornetq-core/src/main/java/org/hornetq/core/protocol/stomp/VersionedStompFrameHandler.java
branches/STOMP11/hornetq-core/src/main/java/org/hornetq/core/protocol/stomp/v11/StompFrameHandlerV11.java
branches/STOMP11/hornetq-core/src/main/java/org/hornetq/core/protocol/stomp/v11/StompFrameV11.java
branches/STOMP11/hornetq-core/src/main/java/org/hornetq/core/server/impl/QueueImpl.java
branches/STOMP11/hornetq-core/src/main/java/org/hornetq/core/server/impl/ServerConsumerImpl.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/StompFrameFactoryV11.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/postoffice/impl/BindingsImpl.java
===================================================================
--- branches/STOMP11/hornetq-core/src/main/java/org/hornetq/core/postoffice/impl/BindingsImpl.java 2011-09-19 04:53:40 UTC (rev 11362)
+++ branches/STOMP11/hornetq-core/src/main/java/org/hornetq/core/postoffice/impl/BindingsImpl.java 2011-09-19 13:17:33 UTC (rev 11363)
@@ -238,15 +238,12 @@
{
if (binding.getFilter() == null || binding.getFilter().match(message))
{
- log.error("---------------------- route to binding: " + binding);
binding.getBindable().route(message, context);
routed = true;
}
}
}
-
- log.error("-------- now routed is: " + routed);
if (!routed)
{
@@ -279,7 +276,6 @@
if (theBinding != null)
{
- log.error("------------------- route theBinding: " + theBinding + " mesage: " + message);
theBinding.route(message, context);
}
}
Modified: branches/STOMP11/hornetq-core/src/main/java/org/hornetq/core/postoffice/impl/PostOfficeImpl.java
===================================================================
--- branches/STOMP11/hornetq-core/src/main/java/org/hornetq/core/postoffice/impl/PostOfficeImpl.java 2011-09-19 04:53:40 UTC (rev 11362)
+++ branches/STOMP11/hornetq-core/src/main/java/org/hornetq/core/postoffice/impl/PostOfficeImpl.java 2011-09-19 13:17:33 UTC (rev 11363)
@@ -591,11 +591,7 @@
cleanupInternalPropertiesBeforeRouting(message);
}
- log.error("----------get address: " + address + " addressManager: " + addressManager);
-
Bindings bindings = addressManager.getBindingsForRoutingAddress(address);
-
- log.error("-------------------Bindings: " + bindings);
if (bindings != null)
{
@@ -634,7 +630,6 @@
}
else
{
- log.error("----------processing route: " + context + " direct " + direct);
processRoute(message, context, direct);
}
@@ -971,8 +966,6 @@
message.incrementRefCount();
}
}
-
- log.error("In processing, tx: " + tx);
if (tx != null)
{
Modified: branches/STOMP11/hornetq-core/src/main/java/org/hornetq/core/protocol/core/impl/ChannelImpl.java
===================================================================
--- branches/STOMP11/hornetq-core/src/main/java/org/hornetq/core/protocol/core/impl/ChannelImpl.java 2011-09-19 04:53:40 UTC (rev 11362)
+++ branches/STOMP11/hornetq-core/src/main/java/org/hornetq/core/protocol/core/impl/ChannelImpl.java 2011-09-19 13:17:33 UTC (rev 11363)
@@ -220,7 +220,6 @@
// The actual send must be outside the lock, or with OIO transport, the write can block if the tcp
// buffer is full, preventing any incoming buffers being handled and blocking failover
- log.error("------------------------ write buffer " + connection);
connection.getTransportConnection().write(buffer, flush, batch);
}
}
Modified: branches/STOMP11/hornetq-core/src/main/java/org/hornetq/core/protocol/core/impl/CoreSessionCallback.java
===================================================================
--- branches/STOMP11/hornetq-core/src/main/java/org/hornetq/core/protocol/core/impl/CoreSessionCallback.java 2011-09-19 04:53:40 UTC (rev 11362)
+++ branches/STOMP11/hornetq-core/src/main/java/org/hornetq/core/protocol/core/impl/CoreSessionCallback.java 2011-09-19 13:17:33 UTC (rev 11363)
@@ -72,8 +72,6 @@
public int sendMessage(ServerMessage message, long consumerID, int deliveryCount)
{
Packet packet = new SessionReceiveMessage(consumerID, message, deliveryCount);
-
- log.error("------------------channel sent " + channel);
channel.sendBatched(packet);
Modified: branches/STOMP11/hornetq-core/src/main/java/org/hornetq/core/protocol/stomp/StompConnection.java
===================================================================
--- branches/STOMP11/hornetq-core/src/main/java/org/hornetq/core/protocol/stomp/StompConnection.java 2011-09-19 04:53:40 UTC (rev 11362)
+++ branches/STOMP11/hornetq-core/src/main/java/org/hornetq/core/protocol/stomp/StompConnection.java 2011-09-19 13:17:33 UTC (rev 11363)
@@ -469,7 +469,6 @@
public void sendFrame(StompFrame frame)
{
- log.error("--------------- sending reply: " + frame);
manager.sendReply(this, frame);
}
@@ -529,9 +528,7 @@
}
try
{
- log.error("--------------------- sending mesage: " + message);
stompSession.getSession().send(message, true);
- log.error("----------------------sent by " + stompSession.getSession());
}
catch (Exception e)
{
Modified: branches/STOMP11/hornetq-core/src/main/java/org/hornetq/core/protocol/stomp/StompDecoder.java
===================================================================
--- branches/STOMP11/hornetq-core/src/main/java/org/hornetq/core/protocol/stomp/StompDecoder.java 2011-09-19 04:53:40 UTC (rev 11362)
+++ branches/STOMP11/hornetq-core/src/main/java/org/hornetq/core/protocol/stomp/StompDecoder.java 2011-09-19 13:17:33 UTC (rev 11363)
@@ -579,8 +579,6 @@
data = data - pos;
// reset
-
- log.error("-------new Frame decoded: " + command + " headers " + headers + " content " + content);
StompFrame ret = new StompFrame(command, headers, content);
Modified: branches/STOMP11/hornetq-core/src/main/java/org/hornetq/core/protocol/stomp/StompFrame.java
===================================================================
--- branches/STOMP11/hornetq-core/src/main/java/org/hornetq/core/protocol/stomp/StompFrame.java 2011-09-19 04:53:40 UTC (rev 11362)
+++ branches/STOMP11/hornetq-core/src/main/java/org/hornetq/core/protocol/stomp/StompFrame.java 2011-09-19 13:17:33 UTC (rev 11363)
@@ -177,15 +177,11 @@
public String getEscapedKey()
{
- log.error("----------------key is : |" + key + "|");
- log.error("----------------esc'd: |" + escape(key) + "|");
return escape(key);
}
public String getEscapedValue()
{
- log.error("----------------val is : |" + val + "|");
- log.error("----------------esc'd v: |" + escape(val) + "|");
return escape(val);
}
Modified: branches/STOMP11/hornetq-core/src/main/java/org/hornetq/core/protocol/stomp/StompSession.java
===================================================================
--- branches/STOMP11/hornetq-core/src/main/java/org/hornetq/core/protocol/stomp/StompSession.java 2011-09-19 04:53:40 UTC (rev 11362)
+++ branches/STOMP11/hornetq-core/src/main/java/org/hornetq/core/protocol/stomp/StompSession.java 2011-09-19 13:17:33 UTC (rev 11363)
@@ -156,6 +156,7 @@
if (sub.getAck().equals(Stomp.Headers.Subscribe.AckModeValues.CLIENT_INDIVIDUAL))
{
+ log.error("---------------------client-individual ack: " + id);
session.individualAcknowledge(consumerID, id);
}
else
Modified: branches/STOMP11/hornetq-core/src/main/java/org/hornetq/core/protocol/stomp/VersionedStompFrameHandler.java
===================================================================
--- branches/STOMP11/hornetq-core/src/main/java/org/hornetq/core/protocol/stomp/VersionedStompFrameHandler.java 2011-09-19 04:53:40 UTC (rev 11362)
+++ branches/STOMP11/hornetq-core/src/main/java/org/hornetq/core/protocol/stomp/VersionedStompFrameHandler.java 2011-09-19 13:17:33 UTC (rev 11363)
@@ -95,14 +95,10 @@
{
response = onUnknown(request.getCommand());
}
-
- log.error("-------------------- handled " + request);
if (response == null)
{
response = postprocess(request);
-
- log.error("---------------postprocessed response: " + response);
}
else
{
Modified: branches/STOMP11/hornetq-core/src/main/java/org/hornetq/core/protocol/stomp/v11/StompFrameHandlerV11.java
===================================================================
--- branches/STOMP11/hornetq-core/src/main/java/org/hornetq/core/protocol/stomp/v11/StompFrameHandlerV11.java 2011-09-19 04:53:40 UTC (rev 11362)
+++ branches/STOMP11/hornetq-core/src/main/java/org/hornetq/core/protocol/stomp/v11/StompFrameHandlerV11.java 2011-09-19 13:17:33 UTC (rev 11363)
@@ -153,7 +153,6 @@
@Override
public StompFrame onDisconnect(StompFrame frame)
{
- log.error("----------------- frame: " + frame);
if (this.heartBeater != null)
{
heartBeater.shutdown();
@@ -447,8 +446,6 @@
StompUtils.copyStandardHeadersFromMessageToFrame(serverMessage, frame, deliveryCount);
- log.error("-------------------- frame created: " + frame);
-
return frame;
}
@@ -456,8 +453,6 @@
@Override
public void replySent(StompFrame reply)
{
- log.error("----------------------- reply sent notified: " + reply);
-
if (reply.getCommand().equals(Stomp.Responses.CONNECTED))
{
//kick off the pinger
@@ -574,8 +569,6 @@
{
dur2 = System.currentTimeMillis() - lastAccepted.get();
- log.error("-------------------------- dur2 is " + dur2);
-
if (dur2 > (2 * serverAcceptPing))
{
connection.disconnect();
@@ -614,21 +607,17 @@
try
{
- log.error("-------------------waiting for " + waitTime);
this.wait(waitTime);
- log.error("--------------------wake up " );
}
catch (InterruptedException e)
{
}
}
- log.error("-------------------------HeartBeat thread shut down!");
}
}
public void pingAccepted()
{
- log.error("------------------------Ping accepted!");
this.lastAccepted.set(System.currentTimeMillis());
}
}
@@ -638,7 +627,6 @@
{
if (heartBeater != null)
{
- log.error("----------------------PING accepted: " + request);
heartBeater.pingAccepted();
}
}
@@ -897,8 +885,6 @@
boolean isEscaping = false;
SimpleBytes holder = new SimpleBytes(1024);
- log.error("--------------------------------- Decoding command: " + decoder.command);
-
outer: while (true)
{
byte b = decoder.workingBuffer[decoder.pos++];
@@ -990,8 +976,6 @@
}
holder.reset();
- log.error("---------- A new header decoded: " + decoder.headerName + " : " + headerValue);
-
decoder.headers.put(decoder.headerName, headerValue);
if (decoder.headerName.equals(StompDecoder.CONTENT_LENGTH_HEADER_NAME))
Modified: branches/STOMP11/hornetq-core/src/main/java/org/hornetq/core/protocol/stomp/v11/StompFrameV11.java
===================================================================
--- branches/STOMP11/hornetq-core/src/main/java/org/hornetq/core/protocol/stomp/v11/StompFrameV11.java 2011-09-19 04:53:40 UTC (rev 11362)
+++ branches/STOMP11/hornetq-core/src/main/java/org/hornetq/core/protocol/stomp/v11/StompFrameV11.java 2011-09-19 13:17:33 UTC (rev 11363)
@@ -71,8 +71,6 @@
}
// Add a newline to separate the headers from the content.
head.append(Stomp.NEWLINE);
-
- log.error("------------------------_______now head: " + head);
buffer.writeBytes(head.toString().getBytes("UTF-8"));
if (bytesBody != null)
Modified: branches/STOMP11/hornetq-core/src/main/java/org/hornetq/core/server/impl/QueueImpl.java
===================================================================
--- branches/STOMP11/hornetq-core/src/main/java/org/hornetq/core/server/impl/QueueImpl.java 2011-09-19 04:53:40 UTC (rev 11362)
+++ branches/STOMP11/hornetq-core/src/main/java/org/hornetq/core/server/impl/QueueImpl.java 2011-09-19 13:17:33 UTC (rev 11363)
@@ -279,7 +279,6 @@
public void route(final ServerMessage message, final RoutingContext context) throws Exception
{
- log.error("-------------------in queue route, context: " + context);
context.addQueue(address, this);
}
@@ -363,7 +362,6 @@
return;
}
- log.error("----------------checkingDirect " + checkDirect);
// The checkDirect flag is periodically set to true, if the delivery is specified as direct then this causes the
// directDeliver flag to be re-computed resulting in direct delivery if the queue is empty
// We don't recompute it on every delivery since executing isEmpty is expensive for a ConcurrentQueue
@@ -386,13 +384,10 @@
checkDirect = false;
}
- log.error("-----now direct " + direct + " directDeliver " + directDeliver );
if (direct && directDeliver && deliverDirect(ref))
{
return;
}
-
- log.error("------- ok, adding ref to the queue");
queueMemorySize.addAndGet(ref.getMessage().getMemoryEstimate());
@@ -401,8 +396,6 @@
directDeliver = false;
executor.execute(concurrentPoller);
-
- log.error("-----------executing : " + concurrentPoller);
}
public void deliverAsync()
@@ -1953,10 +1946,7 @@
HandleStatus status;
try
{
- log.error("-------------------Now let consumer " + consumer + " handle " + reference);
status = consumer.handle(reference);
-
- log.error("-------------- returned status: " + status);
}
catch (Throwable t)
{
Modified: branches/STOMP11/hornetq-core/src/main/java/org/hornetq/core/server/impl/ServerConsumerImpl.java
===================================================================
--- branches/STOMP11/hornetq-core/src/main/java/org/hornetq/core/server/impl/ServerConsumerImpl.java 2011-09-19 04:53:40 UTC (rev 11362)
+++ branches/STOMP11/hornetq-core/src/main/java/org/hornetq/core/server/impl/ServerConsumerImpl.java 2011-09-19 13:17:33 UTC (rev 11363)
@@ -286,7 +286,6 @@
}
else
{
- log.error("--------------------- deliver standard");
deliverStandardMessage(ref, message);
}
@@ -696,7 +695,6 @@
*/
private void deliverStandardMessage(final MessageReference ref, final ServerMessage message)
{
- log.error("------------------ calling callback " + callback + " to send message");
int packetSize = callback.sendMessage(message, id, ref.getDeliveryCount());
if (availableCredits != null)
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-19 04:53:40 UTC (rev 11362)
+++ branches/STOMP11/tests/integration-tests/src/test/java/org/hornetq/tests/integration/stomp/util/AbstractClientStompFrame.java 2011-09-19 13:17:33 UTC (rev 11363)
@@ -72,7 +72,6 @@
String data = new String(sb.toString());
- System.out.println("---------------------------full frame is : " + data);
byte[] byteValue = data.getBytes("UTF-8");
ByteBuffer buffer = ByteBuffer.allocateDirect(byteValue.length);
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-19 04:53:40 UTC (rev 11362)
+++ branches/STOMP11/tests/integration-tests/src/test/java/org/hornetq/tests/integration/stomp/util/AbstractStompClientConnection.java 2011-09-19 13:17:33 UTC (rev 11363)
@@ -164,7 +164,6 @@
while (n >= 0)
{
- System.out.println("read " + n);
if (n > 0)
{
receiveBytes(n);
Modified: branches/STOMP11/tests/integration-tests/src/test/java/org/hornetq/tests/integration/stomp/util/StompFrameFactoryV11.java
===================================================================
--- branches/STOMP11/tests/integration-tests/src/test/java/org/hornetq/tests/integration/stomp/util/StompFrameFactoryV11.java 2011-09-19 04:53:40 UTC (rev 11362)
+++ branches/STOMP11/tests/integration-tests/src/test/java/org/hornetq/tests/integration/stomp/util/StompFrameFactoryV11.java 2011-09-19 13:17:33 UTC (rev 11363)
@@ -46,15 +46,9 @@
public ClientStompFrame createFrame(final String data)
{
- System.out.println("Data: |" + data + "|");
//split the string at "\n\n"
String[] dataFields = data.split("\n\n");
- System.out.println("DataFields[0] |" + dataFields[0]);
- if (dataFields.length > 1)
- {
- System.out.println("DataFields[1] |" + dataFields[1]);
- }
StringTokenizer tokenizer = new StringTokenizer(dataFields[0], "\n");
String command = tokenizer.nextToken();
@@ -63,7 +57,6 @@
while (tokenizer.hasMoreTokens())
{
String header = tokenizer.nextToken();
- System.out.println("header is: " + header);
String[] fields = splitHeader(header);
frame.addHeader(fields[0], fields[1]);
}
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-19 04:53:40 UTC (rev 11362)
+++ branches/STOMP11/tests/integration-tests/src/test/java/org/hornetq/tests/integration/stomp/v11/StompTestV11.java 2011-09-19 13:17:33 UTC (rev 11363)
@@ -21,6 +21,7 @@
import javax.jms.Message;
import javax.jms.MessageConsumer;
+import javax.jms.TextMessage;
import junit.framework.Assert;
@@ -545,6 +546,8 @@
nack(connV11, "sub1", messageID);
+ unsubscribe(connV11, "sub1");
+
connV11.disconnect();
//Nack makes the message be dropped.
@@ -573,6 +576,8 @@
System.out.println("Receiver error: " + error);
+ unsubscribe(connV11, "sub1");
+
connV11.disconnect();
//message should be still there
@@ -601,6 +606,8 @@
System.out.println("Receiver error: " + error);
+ unsubscribe(connV11, "sub1");
+
connV11.disconnect();
//message should still there
@@ -626,6 +633,8 @@
ack(connV11, "sub1", messageID);
+ unsubscribe(connV11, "sub1");
+
connV11.disconnect();
//Nack makes the message be dropped.
@@ -654,6 +663,8 @@
System.out.println("Receiver error: " + error);
+ unsubscribe(connV11, "sub1");
+
connV11.disconnect();
//message should be still there
@@ -682,6 +693,8 @@
System.out.println("Receiver error: " + error);
+ unsubscribe(connV11, "sub1");
+
connV11.disconnect();
//message should still there
@@ -718,6 +731,8 @@
assertEquals("answer-me", error.getHeader("receipt-id"));
+ unsubscribe(connV11, "sub1");
+
connV11.disconnect();
//message should still there
@@ -754,6 +769,8 @@
assertEquals("answer-me", error.getHeader("receipt-id"));
+ unsubscribe(connV11, "sub1");
+
connV11.disconnect();
//message should still there
@@ -786,6 +803,8 @@
//ack the last
this.ack(connV11, "sub1", frame);
+ unsubscribe(connV11, "sub1");
+
connV11.disconnect();
//no messages can be received.
@@ -821,6 +840,8 @@
}
}
+ unsubscribe(connV11, "sub1");
+
connV11.disconnect();
//no messages can be received.
@@ -852,6 +873,8 @@
assertNotNull(frame);
}
+ unsubscribe(connV11, "sub1");
+
connV11.disconnect();
//no messages can be received.
@@ -880,6 +903,7 @@
frame = connV11.receiveFrame();
assertNotNull(frame);
+ System.out.println(i + " == received: " + frame);
//ack on even numbers
if (i%2 == 0)
{
@@ -887,18 +911,23 @@
}
}
+ unsubscribe(connV11, "sub1");
+
connV11.disconnect();
//no messages can be received.
MessageConsumer consumer = session.createConsumer(queue);
- Message message = null;
+ TextMessage message = null;
for (int i = 0; i < num/2; i++)
{
- message = consumer.receive(1000);
+ message = (TextMessage) consumer.receive(1000);
Assert.assertNotNull(message);
+ System.out.println("Legal: " + message.getText());
}
- message = consumer.receive(1000);
+
+ message = (TextMessage) consumer.receive(1000);
+
Assert.assertNull(message);
}
@@ -908,12 +937,15 @@
String messageID = frame.getHeader("message-id");
ClientStompFrame ackFrame = connV11.createFrame("ACK");
- //give it a wrong sub id
+
ackFrame.addHeader("subscription", subId);
ackFrame.addHeader("message-id", messageID);
- ackFrame.addHeader("receipt", "answer-me");
- connV11.sendFrame(ackFrame);
+ ClientStompFrame response = connV11.sendFrame(ackFrame);
+ if (response != null)
+ {
+ throw new IOException("failed to ack " + response);
+ }
}
private void ack(StompClientConnection conn, String subId, String mid) throws IOException, InterruptedException
@@ -944,6 +976,14 @@
conn.sendFrame(subFrame);
}
+ private void unsubscribe(StompClientConnection conn, String subId) throws IOException, InterruptedException
+ {
+ ClientStompFrame subFrame = conn.createFrame("UNSUBSCRIBE");
+ subFrame.addHeader("id", subId);
+
+ conn.sendFrame(subFrame);
+ }
+
}
More information about the hornetq-commits
mailing list