Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

ARTEMIS-856 Test fixes - Alternative #2211

Merged
merged 4 commits into from
Aug 2, 2018
Merged

ARTEMIS-856 Test fixes - Alternative #2211

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
59520b9
ARTEMIS-856 Fixing QueueCommandTest
clebertsuconic Aug 2, 2018
ec742cb
ARTEMIS-856 Fixing MessageRedistributionTest
michaelandrepearce Aug 2, 2018
ddd554f
ARTEMIS-856 Fixing ExclusiveTest
michaelandrepearce Aug 2, 2018
de46517
ARTEMIS-856 Fixing MessageRedistributionTest
michaelandrepearce Aug 2, 2018
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion ...main/java/org/apache/activemq/artemis/core/management/impl/ActiveMQServerControlImpl.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -807,7 +807,7 @@ public String createQueue(String address,
boolean purgeOnNoConsumers,
boolean autoCreateAddress) throws Exception {
AddressSettings addressSettings = server.getAddressSettingsRepository().getMatch(address == null ? name : address);
return createQueue(address, routingType, name, filterStr, durable, addressSettings.getDefaultMaxConsumers(), addressSettings.isDefaultPurgeOnNoConsumers(), addressSettings.isDefaultExclusiveQueue(), addressSettings.isDefaultLastValueQueue(), addressSettings.getDefaultConsumersBeforeDispatch(), addressSettings.getDefaultDelayBeforeDispatch(), addressSettings.isAutoCreateAddresses());
return createQueue(address, routingType, name, filterStr, durable, maxConsumers, purgeOnNoConsumers, addressSettings.isDefaultExclusiveQueue(), addressSettings.isDefaultLastValueQueue(), addressSettings.getDefaultConsumersBeforeDispatch(), addressSettings.getDefaultDelayBeforeDispatch(), autoCreateAddress);
}

@Override
Expand Down
2 changes: 1 addition & 1 deletion ...server/src/main/java/org/apache/activemq/artemis/core/server/impl/ActiveMQServerImpl.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1766,7 +1766,7 @@ public void createSharedQueue(final SimpleString address,
boolean exclusive,
boolean lastValue) throws Exception {
AddressSettings as = getAddressSettingsRepository().getMatch(address == null ? name.toString() : address.toString());
createSharedQueue(address, routingType, name, filterString, user, durable, as.getDefaultMaxConsumers(), as.isDefaultPurgeOnNoConsumers(), as.isDefaultExclusiveQueue(), as.isDefaultLastValueQueue(), as.getDefaultConsumersBeforeDispatch(), as.getDefaultDelayBeforeDispatch());
createSharedQueue(address, routingType, name, filterString, user, durable, maxConsumers, purgeOnNoConsumers, exclusive, lastValue, as.getDefaultConsumersBeforeDispatch(), as.getDefaultDelayBeforeDispatch());
}

@Override
Expand Down
36 changes: 29 additions & 7 deletions artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/QueueImpl.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -21,6 +21,7 @@
import java.math.BigDecimal;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
Expand Down Expand Up @@ -1082,6 +1083,11 @@ private boolean checkConsumerDirectDeliver() {
supports = false;
}
}
if (redistributor != null) {
if (!redistributor.consumer.supportsDirectDelivery()) {
supports = false;
}
}
return supports;
}

Expand Down Expand Up @@ -2158,10 +2164,10 @@ public synchronized int changeReferencesPriority(final Filter filter, final byte
@Override
public synchronized void resetAllIterators() {
for (ConsumerHolder holder : this.consumerList) {
if (holder.iter != null) {
holder.iter.close();
}
holder.iter = null;
holder.resetIterator();
}
if (redistributor != null) {
redistributor.resetIterator();
}
}

Expand Down Expand Up @@ -2484,7 +2490,12 @@ private void deliver() {
}
}

if (pos == endPos) {
if (redistributor != null || groupConsumer != null || exclusive) {
if (noDelivery > 0) {
break;
}
noDelivery = 0;
} else if (pos == endPos) {
// Round robin'd all

if (noDelivery == size) {
Expand Down Expand Up @@ -3040,7 +3051,7 @@ private boolean deliverDirect(final MessageReference ref) {
return true;
}

if (pos == startPos) {
if (pos == startPos || redistributor != null || groupConsumer != null || exclusive) {
// Tried them all
break;
}
Expand Down Expand Up @@ -3122,7 +3133,11 @@ private List<ConsumerHolder> cloneConsumersList() {
List<ConsumerHolder> consumerListClone;

synchronized (this) {
consumerListClone = new ArrayList<>(consumerList);
if (redistributor == null) {
consumerListClone = new ArrayList<>(consumerList);
} else {
consumerListClone = Collections.singletonList(redistributor);
}
}
return consumerListClone;
}
Expand Down Expand Up @@ -3286,6 +3301,13 @@ private static class ConsumerHolder<T extends Consumer> {

LinkedListIterator<MessageReference> iter;

private void resetIterator() {
if (iter != null) {
iter.close();
}
iter = null;
}

}

private class DelayedAddRedistributor implements Runnable {
Expand Down