Is there a way to programmatically delete an ActiveMQ Job Schedule?

Question

I'm trying to delete a scheduled job in ActiveMQ, and so far no luck.

Schedule is created among with the message using either NMS API or Amqpnetlite (except openwire lib as that one is not updated and can't be used on netstandard/netcore)

Sample code used to create the schedule with NMS, same thing done with AMQP lib:

var factory = new Apache.NMS.ActiveMQ.ConnectionFactory(brokerUri);
IConnection connection = factory.CreateConnection(user, password);
connection.Start();

ISession session = connection.CreateSession(AcknowledgementMode.AutoAcknowledge);
IDestination dest = session.GetQueue(destination);
IMessageProducer producer = session.CreateProducer(dest);
producer.DeliveryMode = MsgDeliveryMode.NonPersistent;

var msg = session.CreateTextMessage("Sample text message");
msg.Properties.SetString("AMQ_SCHEDULED_CRON", "* * * * *");
producer.Send(msg);
connection.Close();

This part produces the following result in browser console, and that's what I am willing to delete:

I've read this other question and answer, also active mq system constants but no way the schedule gets deleted. Also tried to browse over the documentation but can't find anything useful so far

Does ActiveMQ even supports a programmative way of managing schedules? An AMQP solution would be great, but NMS is also appreciated.

Answer 1

You can manage scheduled jobs in ActiveMQ over STOMP, AMQP or simply from a JMS client. I've written about this before showing how to do it using the ActiveMQ Java client but the principle is the same. You can send messages with specific headers set that will operate on the scheduled messages.

To browse the collection of scheduled message you'd do something like the following:

    Connection connection = createConnection();
    Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);

    // Create the Browse Destination and the Reply To location
    Destination requestBrowse = session.createTopic(ScheduledMessage.AMQ_SCHEDULER_MANAGEMENT_DESTINATION);
    Destination browseDest = session.createTemporaryQueue();

    // Create the "Browser"
    MessageConsumer browser = session.createConsumer(browseDest);

    connection.start();

    // Send the browse request
    MessageProducer producer = session.createProducer(requestBrowse);
    Message request = session.createMessage();
    request.setStringProperty(ScheduledMessage.AMQ_SCHEDULER_ACTION,
                              ScheduledMessage.AMQ_SCHEDULER_ACTION_BROWSE);
    request.setJMSReplyTo(browseDest);
    producer.send(request);

    Message scheduled = browser.receive(5000);
    while (scheduled != null) {
        // Do something clever...
    }

The returned messages contain the information on the actual scheduled message jobs that have been previously added. Obtaining the job Id allows you to completely cancel the delivery of said message.

To remove a scheduled message send that was scheduled using the Java client, AMQP client or other procotol client you'd do the following:

    Message remove = session.createMessage();
    remove.setStringProperty(ScheduledMessage.AMQ_SCHEDULER_ACTION,
            ScheduledMessage.AMQ_SCHEDULER_ACTION_REMOVE);
    remove.setStringProperty(ScheduledMessage.AMQ_SCHEDULED_ID,
            scheduled.getStringProperty(ScheduledMessage.AMQ_SCHEDULED_ID));
    producer.send(remove);

The full set of message property values that can be used when working with the scheduler are documented here, in AMQP just use the string literal of each as the Application Property value you set with the job Id to remote, or in the NMS client it'd just be a string key message property with the job Id that you want to remove.

There is one caveat though when doing this over AMQP and that would be that you'd need to ensure that the broker is using the JMS transformer ?transport.transformer=jms"see the AMQP documentation for ActiveMQ 5.

Answer 2

Accepted answer is working and correct. Turns out scheduledMessage.NMSMessageId was not holding the scheduler id.

Here's the clean C# source in case one is interested:

var factory = new Apache.NMS.ActiveMQ.ConnectionFactory(brokerUri);
IConnection connection = factory.CreateConnection(userName, password);

var session = connection.CreateSession(AcknowledgementMode.AutoAcknowledge);
var requestBrowse = session.GetTopic("ActiveMQ.Scheduler.Management");
var queue = session.GetQueue(queueName);
var consumer = session.CreateConsumer(queue);

connection.Start();

var producer = session.CreateProducer(requestBrowse);
var scheduledMessage = consumer.Receive(TimeSpan.FromSeconds(10));
if (scheduledMessage != null)
{
    // do check with persistent storage, if schedule is canceled, remove it:
    var remove = session.CreateMessage();
    // get prop names from : http://activemq.apache.org/maven/apidocs/constant-values.htm
    remove.Properties["AMQ_SCHEDULER_ACTION"] = "REMOVE";
    remove.Properties["scheduledJobId"] = scheduledMessage.Properties.GetString("scheduledJobId");
    producer.Send(remove);
}

producer.Close();
session.Close();
connection.Close();

Flow is like this: Get messages from some queue, in case some condition is met, drop the schedule entirely.