CODEWITHSUNDEEP
activemq-classic
Mobility
Mobility

Reputation: 3305

How can I get actievmq queue's size from the "org.apache.activemq.command.Message"?

I want to override send() method of BrokerFilter class.

This method has two params: ProducerBrokerExchange, and Message.

Can I get the destination queue's size from these two params?

I'm using activemq-all-5.13.2.jar.

Upvotes: 0

Views: 638

Answers (2)

greenBat
greenBat

Reputation: 50

You can refer to StatisticsBrokerPlugin.

The code is here if someone else need .

    ActiveMQDestination msgDest = messageSend.getDestination();
    String physicalName = msgDest.getPhysicalName();
    logger.info("dest: " + physicalName);

    ActiveMQDestination queryDestination = ActiveMQDestination.createDestination(physicalName, msgDest.getDestinationType());
    Set<Destination> destinations = getDestinations(queryDestination);
    logger.info(destinations);
    long count = 0;
    for (Destination dest : destinations) {
        DestinationStatistics stats = dest.getDestinationStatistics();
        if (stats != null) {
            count = stats.getMessageSize().getCount();
            logger.info("size: " + count);
        }
    }

Upvotes: 1

Hassen Bennour
Hassen Bennour

Reputation: 3913

you can use : 

BrokerFilter.getRegionDestination().getDestinationStatistics().getMessages().getCount();

you have all this stats in org.apache.activemq.broker.region.DestinationStatistics :

enqueues = new CountStatisticImpl("enqueues", "The number of messages that have been sent to the destination");
dispatched = new CountStatisticImpl("dispatched", "The number of messages that have been dispatched from the destination");
dequeues = new CountStatisticImpl("dequeues", "The number of messages that have been acknowledged from the destination");
forwards = new CountStatisticImpl("forwards", "The number of messages that have been forwarded to a networked broker from the destination");
inflight = new CountStatisticImpl("inflight", "The number of messages dispatched but awaiting acknowledgement");
expired = new CountStatisticImpl("expired", "The number of messages that have expired");

consumers = new CountStatisticImpl("consumers", "The number of consumers that that are subscribing to messages from the destination");
consumers.setDoReset(false);
producers = new CountStatisticImpl("producers", "The number of producers that that are publishing messages to the destination");
producers.setDoReset(false);
messages = new CountStatisticImpl("messages", "The number of messages that that are being held by the destination");
messages.setDoReset(false);
messagesCached = new PollCountStatisticImpl("messagesCached", "The number of messages that are held in the destination's memory cache");
processTime = new TimeStatisticImpl("processTime", "information around length of time messages are held by a destination");
blockedSends = new CountStatisticImpl("blockedSends", "number of messages that have to wait for flow control");
blockedTime = new TimeStatisticImpl("blockedTime","amount of time messages are blocked for flow control");
messageSize = new SizeStatisticImpl("messageSize","Size of messages passing through the destination");

Upvotes: 0

Related Questions