Messaging Pattern 2.4.0 - The Coherence Incubator

This documentation applies to the Messaging Pattern 2.4.0. The latest Messaging Pattern documentation is available here.

The Messaging Pattern

This is an example implementation of store and forward messaging built on top of Coherence.

The Messaging Pattern advocates that;

Payload, typically represented as a Message object, may be sent to a Destination from a Sender (also commonly known as a Publisher).
It is the responsibility of the infrastructure managing the Destination to ensure that Messages (arriving at the said Destination) are then stored (in some manner) and consequently forwarded (in the order in which they arrived at the said Destination) to one or more Receivers (also commonly known as Subscribers).
The Subscribers appropriately consume (receive and acknowledge receipt) of the said Messages from the Destination in the order in which they were forwarded to the said Subscribers.
The infrastructure managing the Messages appropriately clean-up (remove and garbage collect) the said Messages that have been consumed by Subscribers.
The type of the Destination determines the method of delivery to the Subscribers on that Destination.
A Topic Destination (or Topic) will store and forward Messages to all of the Subscribers of the said Topic Destination. This form of Message delivery is often called "publish-and-subscribe messaging", "one-to-many messaging" or "the observer pattern".
A Queue Destination (or Queue) will store and forward Messages to at most one of the Subscribers of the said Queue Destination. For each Message a different Subscriber may be used, but this is implementation and runtime dependent. This form of Message delivery is often called "point-to-point messaging" or "one-to-one messaging".
A Message may be Persistent or Non-Persistent. In the case of Persistent Messages, the infrastructure managing the Destination must safely store the said Messages to a persistent (and recoverable) storage device so that in the case of infrastructure failure, Messages may be recovered (not lost).
A Subscriber to a Topic is either Durable or Non-Durable. Durable Subscriptions allow the system implementing the Subscriber to terminate and return without losing Messages that may have been delivered during the outage (or disconnection).

Outline

Release Name:	Version 2.4.0: July 22nd, 2009
Target Platforms:	Java Standard Edition 5+
Requires Coherence Version:	3.3.1+, 3.4+ or 3.5.2+
Dependencies:	Coherence Common 1.4.0 Command Pattern 2.4.0
Download:	coherence-messagingpattern-2.4.0.jar MD5:3cd9f74a07600255b8449504d124b1d0
Source Code and Documentation:	coherence-messagingpattern-2.4.0-src.zip MD5:48b271a3c370b0a63c39138c37067ff0
Previous Releases:	Messaging Pattern 2.3.0 Messaging Pattern 2.2.0 Messaging Pattern 2.1.1

Dependencies

This project (like other Coherence Incubator projects) uses Apache Ivy for dependency specification and management. While a standard ivy.xml definition file ships with the source and documentation distribution, the following diagram visually indicates the current dependencies.

Rationale

While it's rare that an architecture making extensive use of Coherence will require Store and Forward Messaging (due to the ability to use MapListeners, Continuous Queries and events for notifications), there are arguably some circumstances (when clients may be disconnected) where the pattern is particularly useful.

Although providing an implementation of store and forward Messaging on-top-of-Coherence has always been possible (as demonstrated by this implementation), most application-level requirements for messaging are often satisfied using existing, off-the-shelf and standardized corporate messaging infrastructure, like JMS. In the cases where such infrastructure is not available or not appropriate, this implementation may provide some benefits.

While it is not the intention nor is it the purpose of this implementation to replace existing messaging infrastructure, this implementation is specifically designed to provide a flexible framework for application-specific, high-performance messaging on a Data Grid.

More specifically, this implementation has been designed as a minimal framework to support multi-directional and multi-point (no single point of failure) push replication between multiple Coherence clusters that are deployed and interconnected by high-latency, high-bandwidth but unreliable wide-area-networks (WANs)... aka: The Push Replication Pattern.

Benefits of Datagrid-based Hub-less Messaging

The implementation of the Store and Forward Messaging Pattern on Oracle Coherence provides many unique advantages over traditional hub-and-spoke messaging solutions.

The messaging infrastructure becomes part of and shares the same Coherence Datagrid infrastructure (by default) as your application. Consequently the requirement for separately installed and maintained "messaging servers" is not required.

Being "part" of your application, there is no reason to wrap application objects as Messages (unlike JMS). You can simply publish and consume your application and domain objects.

As all stateful objects in the pattern are managed using standard distributed caches (including Topics, Queues, Subscriptions, and Messages themselves), you may use standard Oracle Coherence features to monitor, manage, query and observe the state of the messaging infrastructure.

Given that the source code is publicly available, you are free to extend and enhance the messaging solution to suit your needs.

Being based on Oracle Coherence, you can rest assured that the solution will provide highly available, scalable (scale-out) and resilient messaging infrastructure.

What's New?

The following changes have been made since the Messaging Pattern 2.3.0 release.

Upgraded to use Coherence Common 1.4.0 and Command Pattern 2.4.0

Introduced namespaces for caches. All cache names used in this project are prefixed with coherence.messagingpattern.

From now on all cache names used by the Incubator patterns have a coherence.<project-name>.* prefixed to ensure uniqueness and avoid cache names clashing with end-user defined application cache names.

Resolves issue where "draining" would not work with in the Push Replication layer PublishingSubscriptions

Allows TopicSubscriptionConfigurations to specify/override the default subscription lease duration

Set the default subscription lease duration to 2 minutes (like a TCP/IP time-out) instead of 10 seconds. This helps solve potential problems when long GC's occur and/or massive re-partitioning delays occur on slow networks.

Resolves issue where non-durable TopicSubscriptions are incorrectly created as Durable TopicSubscriptions

Implementation Discussion

Before you start

The Store and Forward Messaging pattern makes extensive use of the Command Pattern for managing Destinations and Message delivery. In fact the entire messaging infrastructure is modeled as a set of Commands and a few EntryProcessors. Consequently it is advised that you make yourself familiar with the Command Pattern (and its implementation) to understand the implementation of this pattern. Note however that it's know how this pattern is implemented does not preclude you from using it.

Internally all of the standard messaging concepts, like Destinations (Topics and Queues), Subscriptions and Messages, with the exception of Subscribers, are represented as objects placed in one or more Coherence managed distributed caches. Destinations themselves are actually represented as Command Pattern Contexts, which themselves are managed in the "contexts" distributed cache. Almost all messaging interactions, for example creating subscriptions for a Destination, publishing messages to a Destination, are modeled as Command Pattern Commands submitted (or Entry Processors) to the said Destinations for execution. The major consequences and benefits of this approach are that the implementation;

Is naturally geared for accepting as many messages as physically possible (until the network capacity is exceeded or you run out of storage), off-loading the actual delivery to another thread (a CommandExecutor)
Is completely JMX enabled and monitorable. Simply enable Coherence JMX monitoring and the Contexts representing the Destinations will appear in the JMX-tree.
Message delivery is guaranteed to occur in-order (due to the underlying use of the Command Pattern for executing Commands).
Server-loss (i.e.: cluster member loss) does not effect the availability of the messaging infrastructure (as it's built using Coherence distributed caching).

The most significant interface (from an application-level perspective) is the MessagingSession. Implementations of this interface, namely the DefaultMessagingSession, are how you control the messaging infrastructure, including;

Creating Topic and Queue Destinations
Subscribing and Unsubscribing to Destinations (creating Subscribers)
Publishing Payload to Destinations, and
Subscribing and reading (ie: consuming) Payload from Subscriptions

Using Topics and Queues

The following tabs outline the use Topics and Queues in an application.

The following examples outline the use of Topics (one to many messaging), where each published message is delivered to all subscribers.

Step 1: Creating a MessagingSession
Once you've installed and appropriately configured the implementation (see below), the first thing you need to do is create a MessagingSession. This simplest way to achieve this is to use the statically defined DefaultMessagingSession.getInstance() method. (Alternatively you may construct your own instance by providing ContextManager and CommandSubmitter instances to the DefaultMessagingSession constructor)

MessagingSession messagingSession = DefaultMessagingSession.getInstance();

Step 2: Creating a Topic
Once you have a MessagingSession you can create a Topic. As Topics are Command Pattern Contexts, here you have the ability to determine how the underlying infrastructure is configured to manage the submitted Commands to the said Topic. For example, just like registering a Context with the Command Pattern, you may (optionally) provide a ContextConfiguration and Command ManagementStrategy. For example: You can choose how the underlying commands are managed, either colocated with the Topic or distributed across the cluster. For more information, refer to the Command Pattern.

Identifier topicIdentifier = messagingSession.createTopic("my-topic");

or alternatively (specifying a specific configuration)

DefaultContextConfiguration contextConfiguration = new DefaultContextConfiguration(ManagementStrategy.DISTRIBUTED);
  Identifier topicIdentifier = messagingSession.createTopic("my-topic", contextConfiguration);

If the Topic with the specified name (or Identifier) already exists, the Identifier of the existing Topic is returned.

Step 3a: Subscribing to a Topic (non-durable subscribers)
Once you have created a Topic you can subscribe to it by creating a Subscriber. By default, Subscribers to Topics (and Queues) are non-durable. That is, when a Subscriber becomes disconnected from the messaging infrastructure, all of the visible messages for the said Subscriber are appropriately rolled back and the subscription for the Subscriber is removed.

Subscriber subscriber = messagingSession.subscribe(topicIdentifier);

If you don't have the Identifier of the Destination to which you would like to subscribe, you can simply use it's name (as a String). For example;

Subscriber subscriber = messagingSession.subscribe("my-topic");

Step 3b: Subscribing to a Topic (durable subscribers)
To create a durable subscription to a Topic, you need to provide an appropriate TopicSubscriptionConfiguration when subscribing.

Subscriber subscriber = messagingSession.subscribe("my-topic", TopicSubscriptionConfiguration.newDurableConfiguration("my-durable-subscription"));

To resubscribe to a previously created durable subscription, simply call the subscribe method again with the same durable configuration (including the same subscription name)

Step 4: Publishing payload to a Destination

To publish message payload to a Destination, either Topics or Queues, you need to again use the MessagingSession. It's simple, call publishMessage(...) with the Identifier (or name) of the Destination and the corresponding payload.

messagingSession.publishMessage(topicIdentifier, "Hello World");

or

messagingSession.publishMessage("my-topic", "Hello World");

The only thing you need to ensure is that the payload being published is in some way serializable, either by implementing standard Java java.io.serializable or Coherence ExternalizableLite or PortableObject.

Step 5: Consuming a message with a Subscriber
In order to consume a message using a Subscriber you need to use the Subscriber getMessage() method. This method will request, block and wait for a message to be delivered to the Subscriber and then returned to your application.

String message = (String)subscriber.getMessage();

By default newly created Subscribers are in "auto-commit" mode. This means that any message received from a Subscriber will be automatically acknowledged and removed from underlying messaging infrastructure. If however you'd like the opportunity to rollback received messages, simply set the Susbcriber auto commit mode to false. For Example:

subscriber.setAutoCommit(false);

Once you have done this you may make use of the rollback or commit methods on the Subscriber interface to manually control message acknowledgement in the underlying messaging infrastructure.

Like most messaging systems, including JMS, subscribers should only be used by a single thread. If you require concurrent subscriptions to a Destination, simply create more Subcribers.

The following examples outline the use of Queues (many to many messaging), where each message published is delivered to one and only one subscriber.

Step 1: Creating a MessagingSession
Once you've installed and appropriately configured the implementation (see below), the first thing you need to do is create a MessagingSession. This simplest way to achieve this is to use the statically defined DefaultMessagingSession.getInstance() method. (Alternatively you may construct your own instance by providing ContextManager and CommandSubmitter instances to the DefaultMessagingSession constructor)

MessagingSession messagingSession = DefaultMessagingSession.getInstance();

Step 2: Creating a Queue
Much like creating Topics, to create a Queue you use a MessagingSession.

Identifier queueIdentifier = messagingSession.createQueue("my-queue");

or alternatively, like a Topic, you may specifying a configuration.

DefaultContextConfiguration contextConfiguration = new DefaultContextConfiguration(ManagementStrategy.DISTRIBUTED);
  Identifier queueIdentifier = messagingSession.createQueue("my-queue", contextConfiguration);

If the Queue with the specified name (or Identifier) already exists, the Identifier of the existing Queue is returned.

Step 3: Subscribing to a Queue
Once you have created a Queue you can subscribe to it by creating a Subscriber. Subscribers to Queues are always non-durable. That is, when a Subscriber becomes disconnected from the messaging infrastructure, all of the visible messages for the said Subscriber are appropriately rolled back and the subscription for the Subscriber is removed.

Subscriber subscriber = messagingSession.subscribe(queueIdentifier);

If you don't have the Identifier of the Destination to which you would like to subscribe, you can simply use it's name (as a String). For example;

Subscriber subscriber = messagingSession.subscribe("my-queue");

Frequently Asked Questions

Why don't you support the Java Messaging Specification (ie: JMS) or feature X of JMS?

While it is theoretically possible for this implementation to be a SPI (Service Provider Implementation) for the Java Messaging Specification (JMS), this implementation has been explicitly designed to support the development of the Push Replication Pattern and other WAN-based architectures, that of which do not necessarily require JMS.

How can I monitor the infrastructure?

Two ways. Firstly by enabling JMX, you'll find that all Destinations (ie: Topics) are automatically registered into the clustered JMX tree (as ContextExecutors). Secondly, as all of the infrastructure state is represented using Coherence distributed caches; you may examine, listen to and mutate the appropriately named caches, called: "coherence.commandpattern.contexts", "coherence.messagingpattern.messages" and "coherence.messagingpattern.subscriptions".

References and Additional Information

The Coherence Incubator - Command Pattern

Labels