Synchronizing Data Across Mule Applications with MongoDB

One of the many things that Mule does extremely well is transparently store state between messages. Take the idempotent-message-filter for example; The idempotent-message-filter keeps track of what messages have been processed previously to ensure no duplicate messages are processed. This is all achieved with one single xml element. Or take OAuth enabled cloud connectors that need to keep track of access tokens so you're not redirected to some authorize page each time. All this is handled behind the scenes without you having to worry about it.

Mule’s ability to store this state is implemented via object-stores. Object-stores provide a simple generic way to store data in Mule. Many message processors such as the idempotent-meesage-filter, OAuth message processors, until-successful routers and many more all transparently use object-stores behind the scenes to store state between messages.

By default, Mule uses in-memory object-stores behind the scenes. Things get more interesting however, when your Mule application is distributed across multiple Mule nodes. At some point you are going to need to synchronise object-stores across multiple applications. If you are fortunate to be using the Enterprise Edition of Mule you can take advantage of its out of the box clustering support for synchronizing state across an entire cluster. But even then as mentioned in this blog post by John Demic: Synchronizing Mule Applications Across Data Centers with Apache Cassandra, this may not be ideal if your Mule nodes are geographically distributed across multiple data centres.

In the aforementioned post, John gives a great example of how to use Apache Cassandra as a shared object-store. But many other traditonal and NoSql data stores can be used as well. In this blog post I'm going to show a simple example of how you can achieve the same using MongoDB. If it's good enough for the Hipster Hacker then it's good enough for me!

OMG! Does this officially make @MuleSoft Mule part of the Hipster Stack? pic.twitter.com/OwM4dzhGvi

— Ryan Carter (@rydcarter) October 9, 2013

Configuring the MongoDB object-store

MongoDB is an open-source document database, and the leading NoSQL database. The MongoDB module can be used to read/write/query and perform a whole load of other operation a MongoDB database as part of your Mule flows. It also exposes an additional submodule for using your MondogDB database as an object-store implementation in Mule. Here is a sample configuration from one of the tests:

<?xml version="1.0" encoding="UTF-8"?>
<!--

    Mule Mongo Connector

    Copyright (c) MuleSoft, Inc.  All rights reserved.  http://www.mulesoft.com

    The software in this package is published under the terms of the CPAL v1.0
    license, a copy of which has been included with this distribution in the
    LICENSE.txt file.

-->

<mule xmlns="http://www.mulesoft.org/schema/mule/core" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
    xmlns:mos="http://www.mulesoft.org/schema/mule/mongo-object-store"
    xmlns:http="http://www.mulesoft.org/schema/mule/http" xmlns:script="http://www.mulesoft.org/schema/mule/scripting"
    xmlns:test="http://www.mulesoft.org/schema/mule/test" xmlns:spring="http://www.springframework.org/schema/beans"
    xmlns:util="http://www.springframework.org/schema/util" xmlns:p="http://www.springframework.org/schema/p"
    xsi:schemaLocation="
          http://www.mulesoft.org/schema/mule/core http://www.mulesoft.org/schema/mule/core/current/mule.xsd
          http://www.mulesoft.org/schema/mule/mongo-object-store http://www.mulesoft.org/schema/mule/mongo-object-store/current/mule-mongo-object-store.xsd
          http://www.mulesoft.org/schema/mule/http http://www.mulesoft.org/schema/mule/http/current/mule-http.xsd
          http://www.mulesoft.org/schema/mule/scripting http://www.mulesoft.org/schema/mule/scripting/current/mule-scripting.xsd
          http://www.mulesoft.org/schema/mule/test http://www.mulesoft.org/schema/mule/test/current/mule-test.xsd
          http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
          http://www.springframework.org/schema/util http://www.springframework.org/schema/util/spring-util.xsd">

    <spring:beans>
        <spring:bean class="org.mule.module.mongo.FakeObjectStoreUser"
            p:objectStore-ref="mos" />
    </spring:beans>

    <mos:config name="mos" database="mongo-connector-test" />
</mule> 

As you can see, it's classed as it's own module with it's own namespace and is configured simply using the mos:config name="mos" database="mongo-connector-test" element.

However, this doesn't work for the current version of the connector. Due to the following bug. You can simply apply the fix in the aforementioned pull request, or you can instantiate the object store yourself using Spring - similar to that in the Apache Cassandra configuration.

The object-store class can be found here: MongoObjectStore.java. As you can see, it has various annotations for default values and initialising the class. However, as we are not using this as a fully fledged Mule Devkit module, those annotations have no effect. So we have to pass in the default values ourselves and call the initialise method when instantiating our object-store.

This is quite simple with Spring. Here is a full working example using the Spring instantiated object-store as the store for the idempotent-message-filter:

<?xml version="1.0" encoding="UTF-8"?>
<mule xmlns:mongo="http://www.mulesoft.org/schema/mule/mongo" xmlns:objectstore="http://www.mulesoft.org/schema/mule/objectstore"
	xmlns="http://www.mulesoft.org/schema/mule/core"
	xmlns:spring="http://www.springframework.org/schema/beans" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
	xsi:schemaLocation="http://www.mulesoft.org/schema/mule/mongo http://www.mulesoft.org/schema/mule/mongo/2.0/mule-mongo.xsd
		http://www.mulesoft.org/schema/mule/objectstore http://www.mulesoft.org/schema/mule/objectstore/1.0/mule-objectstore.xsd
		http://www.mulesoft.org/schema/mule/core http://www.mulesoft.org/schema/mule/core/current/mule.xsd
		http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-current.xsd">
    
	<spring:beans>
		<spring:bean id="mongoObjectStore" class="org.mule.module.mongo.MongoObjectStore" 
			init-method="initialize" scope="singleton">
			<spring:property name="host" value="localhost"/>
			<spring:property name="port" value="27017"/>
			<spring:property name="database" value="test"/>
			<spring:property name="username" value=""/>
			<spring:property name="password" value=""/>
			<spring:property name="writeConcern" value="DATABASE_DEFAULT"/>
		</spring:bean>
	</spring:beans>
  
	<flow name="mongo-filter">
		<poll frequency="10000">
			<set-payload value="1" doc:name="Set Payload" />
		</poll>
		
		<idempotent-message-filter storePrefix="ids" idExpression="#[payload]">
			<spring-object-store ref="mongoObjectStore" />
		</idempotent-message-filter>
		
		<logger level="INFO" message="Passed Filter" />
	</flow>
	
</mule>

If you run this configuration, Mule will poll every 10 seconds with the exact same message: "1". The first time it polls you should see "Passed Filter" in the log output. Subsequent polls should not log anything as the idempotent message filter will filter them out. This is usually performed with the default in-memory object-store. To prove this is persisted you should be able to check the value in the Mongo database collection: db.getCollection('mule.objectstore._default').find(). Also you can stop your Mule app and restart it to prove that it works between restarts and failovers etc.