mirror of https://github.com/apache/activemq.git
apply fix for: https://issues.apache.org/jira/browse/AMQ-4288
git-svn-id: https://svn.apache.org/repos/asf/activemq/trunk@1441085 13f79535-47bb-0310-9956-ffa450edef68
This commit is contained in:
parent
b8bbca6773
commit
f88b21613b
|
@ -33,6 +33,7 @@ public class OptimizedAckTest extends TestSupport {
|
||||||
private static final Logger LOG = LoggerFactory.getLogger(OptimizedAckTest.class);
|
private static final Logger LOG = LoggerFactory.getLogger(OptimizedAckTest.class);
|
||||||
private ActiveMQConnection connection;
|
private ActiveMQConnection connection;
|
||||||
|
|
||||||
|
@Override
|
||||||
protected void setUp() throws Exception {
|
protected void setUp() throws Exception {
|
||||||
super.setUp();
|
super.setUp();
|
||||||
connection = (ActiveMQConnection) createConnection();
|
connection = (ActiveMQConnection) createConnection();
|
||||||
|
@ -43,6 +44,7 @@ public class OptimizedAckTest extends TestSupport {
|
||||||
connection.setPrefetchPolicy(prefetchPolicy);
|
connection.setPrefetchPolicy(prefetchPolicy);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
@Override
|
||||||
protected void tearDown() throws Exception {
|
protected void tearDown() throws Exception {
|
||||||
connection.close();
|
connection.close();
|
||||||
super.tearDown();
|
super.tearDown();
|
||||||
|
@ -68,13 +70,14 @@ public class OptimizedAckTest extends TestSupport {
|
||||||
}
|
}
|
||||||
}));
|
}));
|
||||||
|
|
||||||
for (int i=0; i<10; i++) {
|
for (int i = 0; i < 10; i++) {
|
||||||
javax.jms.Message msg = consumer.receive(4000);
|
javax.jms.Message msg = consumer.receive(4000);
|
||||||
|
long inFlightCount = regionBroker.getDestinationStatistics().getInflight().getCount();
|
||||||
assertNotNull(msg);
|
assertNotNull(msg);
|
||||||
if (i<7) {
|
if (i < 7) {
|
||||||
assertEquals("all prefetch is still in flight", 10, regionBroker.getDestinationStatistics().getInflight().getCount());
|
assertEquals("all prefetch is still in flight", 10, inFlightCount);
|
||||||
} else {
|
} else {
|
||||||
assertTrue("most are acked but 3 remain", Wait.waitFor(new Wait.Condition(){
|
assertTrue("most are acked but 3 remain", Wait.waitFor(new Wait.Condition() {
|
||||||
@Override
|
@Override
|
||||||
public boolean isSatisified() throws Exception {
|
public boolean isSatisified() throws Exception {
|
||||||
LOG.info("inflight count: " + regionBroker.getDestinationStatistics().getInflight().getCount());
|
LOG.info("inflight count: " + regionBroker.getDestinationStatistics().getInflight().getCount());
|
||||||
|
@ -105,15 +108,15 @@ public class OptimizedAckTest extends TestSupport {
|
||||||
}
|
}
|
||||||
}));
|
}));
|
||||||
|
|
||||||
for (int i=0; i<10; i++) {
|
for (int i = 0; i < 10; i++) {
|
||||||
Thread.sleep(400);
|
Thread.sleep(400);
|
||||||
javax.jms.Message msg = consumer.receive(4000);
|
javax.jms.Message msg = consumer.receive(4000);
|
||||||
long inFlightCount = regionBroker.getDestinationStatistics().getInflight().getCount();
|
long inFlightCount = regionBroker.getDestinationStatistics().getInflight().getCount();
|
||||||
assertNotNull(msg);
|
assertNotNull(msg);
|
||||||
if (i<7) {
|
if (i < 7) {
|
||||||
assertEquals("all prefetch is still in flight: " + i, 10, inFlightCount);
|
assertEquals("all prefetch is still in flight: " + i, 10, inFlightCount);
|
||||||
} else {
|
} else {
|
||||||
assertTrue("most are acked but 3 remain", Wait.waitFor(new Wait.Condition(){
|
assertTrue("most are acked but 3 remain", Wait.waitFor(new Wait.Condition() {
|
||||||
@Override
|
@Override
|
||||||
public boolean isSatisified() throws Exception {
|
public boolean isSatisified() throws Exception {
|
||||||
return 3 == regionBroker.getDestinationStatistics().getInflight().getCount();
|
return 3 == regionBroker.getDestinationStatistics().getInflight().getCount();
|
||||||
|
@ -144,14 +147,14 @@ public class OptimizedAckTest extends TestSupport {
|
||||||
}
|
}
|
||||||
}));
|
}));
|
||||||
|
|
||||||
for (int i=0; i<10; i++) {
|
for (int i = 0; i < 10; i++) {
|
||||||
javax.jms.Message msg = consumer.receive(4000);
|
javax.jms.Message msg = consumer.receive(4000);
|
||||||
long inFlightCount = regionBroker.getDestinationStatistics().getInflight().getCount();
|
long inFlightCount = regionBroker.getDestinationStatistics().getInflight().getCount();
|
||||||
assertNotNull(msg);
|
assertNotNull(msg);
|
||||||
if (i<7) {
|
if (i < 7) {
|
||||||
assertEquals("all prefetch is still in flight", 10, inFlightCount);
|
assertEquals("all prefetch is still in flight", 10, inFlightCount);
|
||||||
} else {
|
} else {
|
||||||
assertTrue("most are acked but 3 remain", Wait.waitFor(new Wait.Condition(){
|
assertTrue("most are acked but 3 remain", Wait.waitFor(new Wait.Condition() {
|
||||||
@Override
|
@Override
|
||||||
public boolean isSatisified() throws Exception {
|
public boolean isSatisified() throws Exception {
|
||||||
LOG.info("inflight count: " + regionBroker.getDestinationStatistics().getInflight().getCount());
|
LOG.info("inflight count: " + regionBroker.getDestinationStatistics().getInflight().getCount());
|
||||||
|
@ -161,7 +164,7 @@ public class OptimizedAckTest extends TestSupport {
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
assertTrue("After delay the scheduled ack should ack all inflight.", Wait.waitFor(new Wait.Condition(){
|
assertTrue("After delay the scheduled ack should ack all inflight.", Wait.waitFor(new Wait.Condition() {
|
||||||
@Override
|
@Override
|
||||||
public boolean isSatisified() throws Exception {
|
public boolean isSatisified() throws Exception {
|
||||||
LOG.info("inflight count: " + regionBroker.getDestinationStatistics().getInflight().getCount());
|
LOG.info("inflight count: " + regionBroker.getDestinationStatistics().getInflight().getCount());
|
||||||
|
|
Loading…
Reference in New Issue