mirror of https://github.com/apache/activemq.git
resolve https://issues.apache.org/activemq/browse/AMQ-2252 - deadlock with nio and optimizeDispatch and test case
git-svn-id: https://svn.apache.org/repos/asf/activemq/trunk@774829 13f79535-47bb-0310-9956-ffa450edef68
This commit is contained in:
parent
34ee8cc7dd
commit
1421317bd2
|
@ -1156,7 +1156,6 @@ public class Queue extends BaseDestination implements Task, UsageListener {
|
||||||
if (!ack.isInTransaction()) {
|
if (!ack.isInTransaction()) {
|
||||||
acknowledge(context, sub, ack, reference);
|
acknowledge(context, sub, ack, reference);
|
||||||
dropMessage(reference);
|
dropMessage(reference);
|
||||||
wakeup();
|
|
||||||
} else {
|
} else {
|
||||||
try {
|
try {
|
||||||
acknowledge(context, sub, ack, reference);
|
acknowledge(context, sub, ack, reference);
|
||||||
|
|
|
@ -0,0 +1,51 @@
|
||||||
|
/**
|
||||||
|
* Licensed to the Apache Software Foundation (ASF) under one or more
|
||||||
|
* contributor license agreements. See the NOTICE file distributed with
|
||||||
|
* this work for additional information regarding copyright ownership.
|
||||||
|
* The ASF licenses this file to You under the Apache License, Version 2.0
|
||||||
|
* (the "License"); you may not use this file except in compliance with
|
||||||
|
* the License. You may obtain a copy of the License at
|
||||||
|
*
|
||||||
|
* http://www.apache.org/licenses/LICENSE-2.0
|
||||||
|
*
|
||||||
|
* Unless required by applicable law or agreed to in writing, software
|
||||||
|
* distributed under the License is distributed on an "AS IS" BASIS,
|
||||||
|
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||||
|
* See the License for the specific language governing permissions and
|
||||||
|
* limitations under the License.
|
||||||
|
*/
|
||||||
|
package org.apache.activemq.broker;
|
||||||
|
|
||||||
|
import java.net.URI;
|
||||||
|
import java.util.ArrayList;
|
||||||
|
import java.util.List;
|
||||||
|
|
||||||
|
import javax.jms.ConnectionFactory;
|
||||||
|
|
||||||
|
import org.apache.activemq.ActiveMQConnectionFactory;
|
||||||
|
import org.apache.activemq.broker.region.policy.PolicyEntry;
|
||||||
|
import org.apache.activemq.broker.region.policy.PolicyMap;
|
||||||
|
|
||||||
|
|
||||||
|
public class NioQueueSubscriptionTest extends QueueSubscriptionTest {
|
||||||
|
|
||||||
|
@Override
|
||||||
|
protected ConnectionFactory createConnectionFactory() throws Exception {
|
||||||
|
return new ActiveMQConnectionFactory("tcp://localhost:62621");
|
||||||
|
}
|
||||||
|
|
||||||
|
@Override
|
||||||
|
protected BrokerService createBroker() throws Exception {
|
||||||
|
BrokerService answer = BrokerFactory.createBroker(new URI("broker://nio://localhost:62621?persistent=false&useJmx=true"));
|
||||||
|
final List<PolicyEntry> policyEntries = new ArrayList<PolicyEntry>();
|
||||||
|
final PolicyEntry entry = new PolicyEntry();
|
||||||
|
entry.setQueue(">");
|
||||||
|
entry.setOptimizedDispatch(true);
|
||||||
|
policyEntries.add(entry);
|
||||||
|
|
||||||
|
final PolicyMap policyMap = new PolicyMap();
|
||||||
|
policyMap.setPolicyEntries(policyEntries);
|
||||||
|
answer.setDestinationPolicy(policyMap);
|
||||||
|
return answer;
|
||||||
|
}
|
||||||
|
}
|
|
@ -303,6 +303,7 @@ public class QueueDuplicatesFromStoreTest extends TestCase {
|
||||||
queue.removeMessage(contextNotInTx, subscription,
|
queue.removeMessage(contextNotInTx, subscription,
|
||||||
new IndirectMessageReference(
|
new IndirectMessageReference(
|
||||||
getMessage(removeIndex)), ack);
|
getMessage(removeIndex)), ack);
|
||||||
|
queue.wakeup();
|
||||||
|
|
||||||
}
|
}
|
||||||
if (removeIndex % 1000 == 0) {
|
if (removeIndex % 1000 == 0) {
|
||||||
|
|
Loading…
Reference in New Issue