Author: dejanb
Date: Tue Dec 16 03:02:28 2008
New Revision: 727017
URL: http://svn.apache.org/viewvc?rev=727017&view=rev
Log:
fix for prefetch size issue reported in AMQ-1807
Modified:
activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/PrefetchSubscription.java
activemq/trunk/activemq-core/src/main/java/org/apache/activemq/transport/stomp/StompConnection.java
activemq/trunk/activemq-core/src/test/java/org/apache/activemq/transport/stomp/StompTest.java
Modified:
activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/PrefetchSubscription.java
URL:
http://svn.apache.org/viewvc/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/PrefetchSubscription.java?rev=727017&r1=727016&r2=727017&view=diff
==============================================================================
---
activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/PrefetchSubscription.java
(original)
+++
activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/PrefetchSubscription.java
Tue Dec 16 03:02:28 2008
@@ -243,11 +243,11 @@
// consumer
if (getPrefetchSize() != 0) {
prefetchExtension = Math.max(
- prefetchExtension, index + 1);
+ prefetchExtension, index );
}
} else {
prefetchExtension = Math.max(0,
- prefetchExtension - (index + 1));
+ prefetchExtension - index);
}
destination = node.getRegionDestination();
callDispatchMatched = true;
Modified:
activemq/trunk/activemq-core/src/main/java/org/apache/activemq/transport/stomp/StompConnection.java
URL:
http://svn.apache.org/viewvc/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/transport/stomp/StompConnection.java?rev=727017&r1=727016&r2=727017&view=diff
==============================================================================
---
activemq/trunk/activemq-core/src/main/java/org/apache/activemq/transport/stomp/StompConnection.java
(original)
+++
activemq/trunk/activemq-core/src/main/java/org/apache/activemq/transport/stomp/StompConnection.java
Tue Dec 16 03:02:28 2008
@@ -112,6 +112,11 @@
headers.put("passcode", password);
StompFrame frame = new StompFrame("CONNECT", headers);
sendFrame(frame.toString());
+
+ StompFrame connect = receive();
+ if (!connect.getAction().equals(Stomp.Responses.CONNECTED)) {
+ throw new Exception ("Not connected: " + connect.getBody());
+ }
}
public void disconnect() throws Exception {
Modified:
activemq/trunk/activemq-core/src/test/java/org/apache/activemq/transport/stomp/StompTest.java
URL:
http://svn.apache.org/viewvc/activemq/trunk/activemq-core/src/test/java/org/apache/activemq/transport/stomp/StompTest.java?rev=727017&r1=727016&r2=727017&view=diff
==============================================================================
---
activemq/trunk/activemq-core/src/test/java/org/apache/activemq/transport/stomp/StompTest.java
(original)
+++
activemq/trunk/activemq-core/src/test/java/org/apache/activemq/transport/stomp/StompTest.java
Tue Dec 16 03:02:28 2008
@@ -884,7 +884,7 @@
stompConnection.sendFrame(frame);
// wait a bit for MBean to get refreshed
try {
- Thread.sleep(100);
+ Thread.sleep(200);
} catch (InterruptedException e){}
assertEquals(view.getDurableTopicSubscribers().length, 1);
@@ -892,7 +892,7 @@
frame = "DISCONNECT\nclient-id:test\n\n" + Stomp.NULL;
stompConnection.sendFrame(frame);
try {
- Thread.sleep(100);
+ Thread.sleep(200);
} catch (InterruptedException e){}
//reconnect
@@ -920,17 +920,41 @@
stompConnection.begin("tx1");
stompConnection.send("/queue/" + getQueueName(), "msg", "tx1", null);
stompConnection.commit("tx1");
-
- StompFrame connect = stompConnection.receive();
- if (!connect.getAction().equals(Stomp.Responses.CONNECTED)) {
- throw new Exception ("Not connected");
- }
stompConnection.subscribe("/queue/" + getQueueName());
StompFrame stompMessage = stompConnection.receive();
assertNull(stompMessage.getHeaders().get("transaction"));
}
+ public void testPrefetchSize() throws Exception {
+ stompConnection.connect("system", "manager");
+
+ HashMap<String, String> headers = new HashMap<String, String>();
+ headers.put("activemq.prefetchSize", "1");
+ stompConnection.subscribe("/queue/" + getQueueName(), "client",
headers);
+
+ // send messages using JMS
+ sendMessage("message 1");
+ sendMessage("message 2");
+ sendMessage("message 3");
+
+ StompFrame frame = stompConnection.receive();
+
+ stompConnection.begin("tx1");
+ stompConnection.ack(frame, "tx1");
+
+ StompFrame frame1 = stompConnection.receive();
+
+ try {
+ StompFrame frame2 = stompConnection.receive(500);
+ if (frame2 != null) {
+ fail("Should not have received the second message");
+ }
+ } catch (SocketTimeoutException soe) {}
+ stompDisconnect();
+
+ }
+
protected void assertClients(int expected) throws Exception {
org.apache.activemq.broker.Connection[] clients =
broker.getBroker().getClients();
int actual = clients.length;
