Apache
/
activemq-artemis
mirror of https://github.com/apache/activemq-artemis.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

ARTEMIS-1961 track routed and unrouted messages sent to an address

This commit is contained in:
Justin Bertram 2018-06-29 09:19:15 -05:00 committed by Clebert Suconic
parent a9916adcb4
commit 9c62531c2f
6 changed files with 102 additions and 14 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

12
artemis-core-client/src/main/java/org/apache/activemq/artemis/api/core/management/AddressControl.java
View File

@ -104,6 +104,18 @@ public interface AddressControl {
@Attribute(desc = "number of messages added to all the queues for this address")
long getMessageCount();
/**
* Returns the number of messages routed to one or more bindings
*/
@Attribute(desc = "number of messages routed to one or more bindings")
long getRoutedMessageCount();
/**
* Returns the number of messages not routed to any bindings
*/
@Attribute(desc = "number of messages not routed to any bindings")
long getUnRoutedMessageCount();
/**
* @param headers the message headers and properties to set. Can only

10
artemis-server/src/main/java/org/apache/activemq/artemis/core/management/impl/AddressControlImpl.java
View File

@ -267,6 +267,16 @@ public class AddressControlImpl extends AbstractControl implements AddressContro
return getMessageCount(DurabilityType.ALL);
}
@Override
public long getRoutedMessageCount() {
return addressInfo.getRoutedMessageCount();
}
@Override
public long getUnRoutedMessageCount() {
return addressInfo.getUnRoutedMessageCount();
}
@Override
public String sendMessage(final Map<String, String> headers,

29
artemis-server/src/main/java/org/apache/activemq/artemis/core/postoffice/impl/PostOfficeImpl.java
View File

@ -842,12 +842,12 @@ public class PostOfficeImpl implements PostOffice, NotificationListener, Binding
throw new IllegalStateException("Message cannot be routed more than once");
}
setPagingStore(context.getAddress(message), message);
final SimpleString address = context.getAddress(message);
setPagingStore(address, message);
AtomicBoolean startedTX = new AtomicBoolean(false);
final SimpleString address = context.getAddress(message);
applyExpiryDelay(message, address);
if (!checkDuplicateID(message, context, rejectDuplicates, startedTX)) {
@ -856,23 +856,24 @@ public class PostOfficeImpl implements PostOffice, NotificationListener, Binding
message.cleanupInternalProperties();
Bindings bindings = addressManager.getBindingsForRoutingAddress(context.getAddress(message));
Bindings bindings = addressManager.getBindingsForRoutingAddress(address);
AddressInfo addressInfo = addressManager.getAddressInfo(address);
// TODO auto-create queues here?
// first check for the auto-queue creation thing
if (bindings == null) {
// There is no queue with this address, we will check if it needs to be created
// if (queueCreator.create(address)) {
// TODO: this is not working!!!!
// reassign bindings if it was created
// bindings = addressManager.getBindingsForRoutingAddress(address);
// }
}
if (bindingMove != null) {
bindingMove.route(message, context);
if (addressInfo != null) {
addressInfo.incrementRoutedMessageCount();
}
} else if (bindings != null) {
bindings.route(message, context);
if (addressInfo != null) {
addressInfo.incrementRoutedMessageCount();
}
} else {
if (addressInfo != null) {
addressInfo.incrementUnRoutedMessageCount();
}
// this is a debug and not warn because this could be a regular scenario on publish-subscribe queues (or topic subscriptions on JMS)
if (logger.isDebugEnabled()) {
logger.debug("Couldn't find any bindings for address=" + address + " on message=" + message);

25
artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/AddressInfo.java
View File

@ -22,6 +22,7 @@ import org.apache.activemq.artemis.utils.PrefixUtil;
import java.util.EnumSet;
import java.util.Map;
import java.util.concurrent.atomic.AtomicLongFieldUpdater;
public class AddressInfo {
@ -36,6 +37,14 @@ public class AddressInfo {
private boolean internal = false;
private volatile long routedMessageCount = 0;
private static final AtomicLongFieldUpdater<AddressInfo> routedMessageCountUpdater = AtomicLongFieldUpdater.newUpdater(AddressInfo.class, "routedMessageCount");
private volatile long unRoutedMessageCount = 0;
private static final AtomicLongFieldUpdater<AddressInfo> unRoutedMessageCountUpdater = AtomicLongFieldUpdater.newUpdater(AddressInfo.class, "unRoutedMessageCount");
public AddressInfo(SimpleString name) {
this(name, EnumSet.noneOf(RoutingType.class));
}
@ -155,4 +164,20 @@ public class AddressInfo {
return this;
}
public long incrementRoutedMessageCount() {
return routedMessageCountUpdater.incrementAndGet(this);
}
public long incrementUnRoutedMessageCount() {
return unRoutedMessageCountUpdater.incrementAndGet(this);
}
public long getRoutedMessageCount() {
return routedMessageCountUpdater.get(this);
}
public long getUnRoutedMessageCount() {
return unRoutedMessageCountUpdater.get(this);
}
}

30
tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/management/AddressControlTest.java
View File

@ -340,6 +340,36 @@ public class AddressControlTest extends ManagementTestBase {
assertTrue(Wait.waitFor(() -> addressControl.getMessageCount() == 2, 2000, 100));
}
@Test
public void testGetRoutedMessageCounts() throws Exception {
SimpleString address = RandomUtil.randomSimpleString();
session.createAddress(address, RoutingType.ANYCAST, false);
AddressControl addressControl = createManagementControl(address);
assertEquals(0, addressControl.getMessageCount());
ClientProducer producer = session.createProducer(address.toString());
producer.send(session.createMessage(false));
assertTrue(Wait.waitFor(() -> addressControl.getRoutedMessageCount() == 0, 2000, 100));
assertTrue(Wait.waitFor(() -> addressControl.getUnRoutedMessageCount() == 1, 2000, 100));
session.createQueue(address, RoutingType.ANYCAST, address);
producer.send(session.createMessage(false));
assertTrue(Wait.waitFor(() -> addressControl.getRoutedMessageCount() == 1, 2000, 100));
assertTrue(Wait.waitFor(() -> addressControl.getUnRoutedMessageCount() == 1, 2000, 100));
session.createQueue(address, RoutingType.ANYCAST, address.concat('2'));
producer.send(session.createMessage(false));
assertTrue(Wait.waitFor(() -> addressControl.getRoutedMessageCount() == 2, 2000, 100));
assertTrue(Wait.waitFor(() -> addressControl.getUnRoutedMessageCount() == 1, 2000, 100));
session.deleteQueue(address);
session.deleteQueue(address.concat('2'));
producer.send(session.createMessage(false));
assertTrue(Wait.waitFor(() -> addressControl.getRoutedMessageCount() == 2, 2000, 100));
assertTrue(Wait.waitFor(() -> addressControl.getUnRoutedMessageCount() == 2, 2000, 100));
}
@Test
public void testSendMessage() throws Exception {
SimpleString address = RandomUtil.randomSimpleString();

10
tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/management/AddressControlUsingCoreTest.java
View File

@ -103,6 +103,16 @@ public class AddressControlUsingCoreTest extends AddressControlTest {
return (long) proxy.retrieveAttributeValue("messageCount");
}
@Override
public long getRoutedMessageCount() {
return (long) proxy.retrieveAttributeValue("routedMessageCount");
}
@Override
public long getUnRoutedMessageCount() {
return (long) proxy.retrieveAttributeValue("unRoutedMessageCount");
}
@Override
public String sendMessage(Map<String, String> headers,
int type,