Manual Acknowledgement

60East generally recommends that applications use an ack() method to acknowledge messages during normal processing. This approach works properly from within a message handler, provides batching support as described elsewhere in this chapter, and is generally both easier to code and more efficient.

However, in some situations, you may need to manually acknowledge messages in the queue. This is most common when an application needs to operate on all messages with certain characteristics, rather than acknowledging individual messages. For example, an application that is doing updates to an order may want cancel an order by both publishing a cancellation and immediately expiring all other messages in the queue for that order. With manual acknowledgment, that application can use a filter to remove all previous updates for that order, then publish the cancellation.

To manually acknowledge processed messages and remove the messages from the queue, applications use the sow_delete command. To remove specific messages from the queue, provide the bookmarks of those messages. To remove messages that match a given filter, provide the filter. Notice that AMPS only supports using a bookmark with sow_delete when removing messages from a queue, not when removing records from a SOW.

For example, given a Message object to acknowledge and a client, the code below acknowledges the message.

void acknowledgeSingle(const Client& client, const Message& message)
{
    Command acknowledge("sow_delete");
    acknowledge.setTopic(message.getTopic())
               .setBookmark(message.getBookmark());
    client.executeAsync(acknowledge, MessageHandler());
}

In the above listing the program creates a sow_delete command, specifies the topic and the bookmark, and then sends the command to the server.

While this method works, creating and sending an acknowledgment for each individual message can be inefficient if your application is processing a large volume of messages. Rather than acknowledging each message individually, your application can build a comma-delimited list of bookmarks from the processed messages and acknowledge all of the messages at the same time. In this case, it's important to be sure that the number of messages you wait for is less than the maximum backlog -- the number of messages your client can have unacknowledged at a given time. Notice that both automatic acknowledgment and the helper method on the Message object take the maximum backlog into account.

When constructing a command to acknowledge queue messages, AMPS allows an application to specify a filter rather than a set of bookmarks. AMPS interprets this as the client requesting acknowledgment of all messages that match the filter. (This may include messages that the client has not received, subject to the Leasing model for the queue.)

As a more typical example of manual acknowledgment, the code below expires all messages for a given id that have a status other than cancel. An application might do this to halt processing of an order that it is about to cancel:

void removePending(const Client& client,  const std::string& orderId)
{
    Command acknowledge("sow_delete");
    acknowledge.setTopic(message.getTopic())
               .setFilter("/id = '" + orderId +"' and /status != 'cancel'")
               .setOptions("expire");
    client.executeAsync(acknowledge, MessageHandler());
}

In the above listing the program specifies a topic and a filter to use to find the messages that should be removed. In this case, the program also provides the expire option to indicate that the messages have been removed from the queue rather than successfully processed (of course, whether this is the correct behavior for a canceled order depends on the expected message flow for your application).

Notice that, as described in the section on multithreading, this method of acknowledging a message should not be used from a message handler unless the sow_delete is sent from a different client than the client that called the message handler. Instead, 60East recommends using the ack() function from within a message handler.