MongoDB Transport

I just released a Mule transport for MongoDB. Full details, documentation and download links are here.

jmxsh

I recently discovered jmxsh, which provides TCL scripting facilities to interact with JMX. Its pretty impressive when compared to something like jconsole. For instance, I was able to write the following little script to query HornetQ for the amount of messages on a queue:

set host  [lindex $argv 0]
set port  [lindex $argv 1]
set queue [lindex $argv 2]

jmx_connect -h $host -p $port
set message_count [jmx_get  -m org.hornetq:module=JMS,name="$queue",type=Queue MessageCount]

puts "$message_count"

jmx_close

I can then run the script like this and it prints the amount of messages on the DLQ to stdout:

./jmxsh queue_count.tcl localhost 3000 DLQ

While the use of TCL might seem obtuse (ie, why not Groovy?), it makes sense from the standpoint of a sysadmin. The JMX agnostic language allows them to script against an app's MBeans with minimal exposure to Java or the JMX API's.

Component Implementation Guidelines

I've recently worked on a couple Mule ESB projects that make heavy use of components. The projects were similar in the sense that the component logic was implemented from scratch. We were not, for instance, exposing existing Java classes out over a transport.

It was challenging at times to implement these components while maintaining best-practices with decoupling the code from Mule and unit-testing it. The below are some approaches / guidelines when implementing components that keep these goals in mind:

• Be careful when implementing Callable. Implementing the Callable interface gives you access to the MuleEventContext when the "onCall" method is invoked. This allows you to invoke the endpoint's transformers, access the MuleMessage directly, stop message processing, etc. It also tightly couples your component code to the Mule infrastructure, making the component more difficult to unit-test in isolation as well as refactor if the Mule API changes. See if your component's reliance on the MuleEventContext can be refactored out to transformers of exception-strategies.
• Favor using a canonical data model over interacting with MuleMessage directly. You can use a transformer to move the external data into a common format, like a domain model class, an XML document or a hash-map. This again decouples your code from the Mule API, making unit-testing, refactoring, etc easier.
• Hide MuleClient usage behind a service interface. MuleClient is an extremely useful facility to send and receive messages over arbitrary transports. Its usage introduces the same difficulties as the items above. This can be mitigated by abstracting the MuleClient invocations into a service class that can be injected into the component. A mock implementation of this class can then be used in unit tests.
• Consider using jBPM to orchestrate activity and maintain state. Component code, along with MuleClient, can be a tempting place to orchestrate, compose and maintain state across endpoint invocations. But be careful when considering this approach. You'll need to consider what happens if the component logic is interrupted during execution, if the state needs to be transient across Mule restarts, if the state needs to be transient between Mule nodes, etc. Many, if not most, of these issues can be addressed by using jBPM in conjunction with Mule's BPM transport.
• Avoid using hardcoded strings in endpoint names with MuleClient. This is obvious but is worth mentioning. Try to centralize all the endpoint names in one place (in an enum, for instance) and inject this into your service facades. This helps avoid issues with typos in endpoint addresses.

Hopefully some of the above will give you component implementation / configurations that are easy to unit-test and refactor. Any other tips / guidelines are welcomed in the comments :)

Mule and JMS as Decoupling Middleware

I just finished reading Michael Nygard's excellent "Release It!". His war stories echo a lot of my own experience, particularly when I was doing ops work full-time. Its a definite must read for developers and system administrators alike.

In Chapter 5 Michael describes the "Decoupling Middleware" pattern. He suggests using a messaging broker to decouple remote service invocation. This allows you to leverage the features of the messaging broker, like durability and delayed re-delivery, to improve the resiliency of communication with a remote service.

The following demonstrates how to use Mule and JMS to decouple interaction with Twitter's API. "Tweet Service"
transactionally accepts messages from a JMS queue and submits them to Twitter's REST API:

<model name="Twitter Services">

  <service name="Tweet Service">
      <inbound>             
          <jms:inbound-endpoint queue="tweets">
              <jms:transaction action="BEGIN_OR_JOIN"/>
          </jms:inbound-endpoint>
      </inbound>
      <http:rest-service-component
                   httpMethod="POST"
                   serviceUrl="http://user:password@twitter.com/statuses/update.json">
          <http:requiredParameter key="status" value="#[payload:]"/>
      </http:rest-service-component>
   </service>
</model>

Let's consider some of the benefits of this indirection:

• The service can be taken down / brought up without fear of losing messages - they will queue on the broker and be delivered when the service is brought back up.


• The service can be scaled horizontally simply by adding additional instances (competing consumption off the queue)


• Messages can be re-ordered based on priority and sent to the remote service


• Requests can be re-submitted in the event of a failure.

The last point is particularly important. Its common to encounter transient errors when integrating with remote services, particularly web-based ones. These errors are usually recoverable after a certain amount of time. If your JMS broker supports it, you can use delayed redelivery to periodically re-attempt the request. HornetQ supports this by configuring address-settings on the queue. The following address-settings for the "tweets" queue specifies 12 redelivery attempts with a 5 minute interval between each request:

<address-setting match="jms.queue.tweets">
   <max-delivery-attempts>12</max-delivery-attempts>
   <redelivery-delay>300000</redelivery-delay>
</address-setting>

Its incidentally easy to employ a "circuit-breaker", another one of Michael's patterns, into the above using an exception-strategy. I'll demonstrate that in another post.

HornetQ and Mule

The lack of OpenMQ's delivery retry options have led me down the road of evaluating JMS brokers again. I luckily didn't have to look very far. JBoss' recently released HornetQ broker is very impressive. Getting it going with Mule is trivial. Here's the config I'm using, which connects to a local HornetQ instance using Netty (instead of JNDI.)

   <spring:bean name="transportConfiguration"
               class="org.hornetq.core.config.TransportConfiguration">
       <spring:constructor-arg
       value="org.hornetq.integration.transports.netty.NettyConnectorFactory"/>
   </spring:bean>

   <spring:bean name="connectionFactory"
       class="org.hornetq.jms.client.HornetQConnectionFactory">
      <spring:constructor-arg ref="transportConfiguration"/>
      <spring:property name="minLargeMessageSize" value="250000"/>
      <spring:property name="cacheLargeMessagesClient" value="false"/>
   </spring:bean>

   <jms:connector name="jmsConnector"
                  connectionFactory-ref="connectionFactory"
                  createMultipleTransactedReceivers="false"
                  numberOfConcurrentTransactedReceivers="1"
                   specification="1.1">
   </jms:connector>

Mule Endpoint QoS with Esper

Esper is a really slick, open-source CEP engine I've been playing with to monitor traffic on Mule endpoints.  Monitoring endpoints with port checks, JMX and log monitoring gives a lot of insight into the health of individual Mule instances, but offers little insight when external services fail.  An external producer of JMS messages to a queue may fail, a database may have a slow-query situation where rows take longer then expected to return or an SMTP outage may stop messages from being delivered to an IMAP server.  Any of these situations would cause less then an expected amount of messages to be delivered to a JMS, JDBC or IMAP endpoint.

By using a wiretap-router or envelope-interceptor on an inbound-endpoint, data about incoming messages can be sent to a CEP engine to construct an event stream.  A query can then be written that produces an event when less messages are seen then expected on the stream.  

A quick demonstration of this follows.  Here are a couple of Groovy scripts that will be wired up with Spring and used to monitor a CXF inbound-endpoint on a Mule instance with Esper.

import org.mule.api.lifecycle.Callable
import org.mule.api.MuleEventContext

class EventInjector implements Callable {

     def esperService

     public Object onCall(MuleEventContext context) {
         esperService.getEPRuntime().sendEvent(context.getMessage())
     }

}

This component will be used to receive messages off the wiretap and inject them into the event stream.  The next component will be used to register listeners on the stream.

import com.espertech.esper.client.UpdateListener
import com.espertech.esper.client.EventBean

import org.mule.module.client.MuleClient

class MuleEventListener implements UpdateListener {

    def expression
    def payloadExpression
    def esperService
    def endpoint

    def initialize() {
        def statement = esperService.getEPAdministrator().createEPL(expression);
        statement.addListener(this)
    }

    public void update(EventBean[] newEvents, EventBean[] oldEvents) {
        def client = new MuleClient()
        def event = newEvents[0];
        client.dispatch(endpoint, event.get(payloadExpression), null)
     }

}

This component code takes two Esper expressions.  expression queries the event stream for events.  payloadExpression populates the message payload of the new message.  endpoint is where this message will be published to.  Here is the Spring beans config that wires the two component scripts with Esper.

<beans xmlns="http://www.springframework.org/schema/beans"
       xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
       xmlns:lang="http://www.springframework.org/schema/lang"
       xsi:schemaLocation=
               "http://www.springframework.org/schema/beans
                http://www.springframework.org/schema/beans/spring-beans-2.5.xsd
                http://www.springframework.org/schema/lang
                http://www.springframework.org/schema/lang/spring-lang-2.5.xsd
                ">

    <bean id="esperService" scope="singleton"
          class="com.espertech.esper.client.EPServiceProviderManager"
          factory-method="getDefaultProvider"/>

    <lang:groovy id="eventInjector"
                 script-source="classpath:/EventInjector.groovy">
        <lang:property name="esperService" ref="esperService"/>
    </lang:groovy>

    <lang:groovy id="mininumMessageListener"
                 script-source="classpath:/MuleEventListener.groovy"
                 init-method="initialize">
        <lang:property name="esperService" ref="esperService"/>
        <lang:property name="endpoint" value="jms://topic:alerts"/>
        <lang:property name="expression" 
                       value="select count(*) from org.mule.api.MuleMessage.win:time_batch(10, 'FORCE_UPDATE, START_EAGER') having count(*) < 5"/>
        <lang:property name="payloadExpression" value="count(*)"/>
    </lang:groovy>
</beans>

"mininumMessageListener" will send a JMS message to the "alerts" topic when less then 5 messages appear on the stream in a 10 second window.  The following Mule config pulls all the above together.

<mule xmlns="http://www.mulesource.org/schema/mule/core/2.2"
    xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
    xmlns:spring="http://www.springframework.org/schema/beans"
    xmlns:test="http://www.mulesource.org/schema/mule/test/2.2"
    xmlns:jms="http://www.mulesource.org/schema/mule/jms/2.2"
    xmlns:vm="http://www.mulesource.org/schema/mule/vm/2.2"
    xmlns:cxf="http://www.mulesource.org/schema/mule/cxf/2.2"
    xmlns:mule-xml="http://www.mulesource.org/schema/mule/xml/2.2"
    xsi:schemaLocation="
    http://www.mulesource.org/schema/mule/core/2.2 http://www.mulesource.org/schema/mule/core/2.2/mule.xsd
    http://www.mulesource.org/schema/mule/cxf/2.2 http://www.mulesource.org/schema/mule/cxf/2.2/mule-cxf.xsd
    http://www.mulesource.org/schema/mule/test/2.2 http://www.mulesource.org/schema/mule/test/2.2/mule-test.xsd
    http://www.mulesource.org/schema/mule/jms/2.2 http://www.mulesource.org/schema/mule/jms/2.2/mule-jms.xsd
    http://www.mulesource.org/schema/mule/vm/2.2 http://www.mulesource.org/schema/mule/vm/2.2/mule-vm.xsd
    http://www.mulesource.org/schema/mule/xml/2.2 http://www.mulesource.org/schema/mule/xml/2.2/mule-xml.xsd
    http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-2.5.xsd">


<spring:beans>
<spring:import resource="classpath:spring-config.xml"/>
</spring:beans>

<vm:connector name="vmConnector" queueEvents="true"/>

<cxf:connector name="cxfConnector"/>

<model name="main">

<service name="soapService">
    <inbound>
        <cxf:inbound-endpoint address="http://localhost:9756/people" connector-ref="cxfConnector"
                              frontend="simple"
                              serviceClass="org.mule.tck.testmodels.services.PeopleService"/>
        <wire-tap-router>
            <vm:outbound-endpoint path="cep.in"/>
        </wire-tap-router>
    </inbound>
    <test:web-service-component/>
</service>

<service name="Complex Event Processing Service">
    <inbound>
        <vm:inbound-endpoint path="cep.in"/>
    </inbound>
    <component>
        <spring-object bean="eventInjector"/>
    </component>
</service>

</model>

</mule>

This example is simplistic, but hopefully the usefulness of this sort of approach is obvious.  One particular improvement is to use JMS instead of VM as the target of the wiretap.  In this scenario, "Complex Event Processing Service" could be hosted in a separate Mule instance dedicated for event analysis. This would additionally allow horizontally load-balancing "soapService" instances to contribute to the same event stream.

I'm additionally using the MuleMessage as the event type.  This offers a limited view into the messages.  A more useful implementation would operate on the payload of the messages, via Maps, POJO's or XML.  The online Esper documentation is extremely well-written and offers examples to get that going.  

SAAJ Transport -> SAAJ Module

With the time I had off work, editing the book and (briefly from) feeding the baby a few weeks ago, I naturally decided to add receiver functionality to the SAAJ transport. In a sleep deprived state  I considered extending the HTTP and JMS transports to use SAAJ to receive and dispatch messages. After chatting with David, however, a cleaner approach seemed to be in order. I started to work on refactoring the code I was using in the SAAJ MessageAdapter, which moved the message payloads back and forth from SOAP, into dedicated transformer implementations: soap-message-to-document transformer and document-to-soap-message-transformer.  

The refactored implementation allows you to use the SAAJ transformers to transform message payloads over arbitrary transports, like VM, XMPP or file, in addition to HTTP and JMS. The transformers are available from the (renamed) SAAJ Module. Cursory documentation and a few examples are available.  The distribution link isn't working yet, but you can download the jars and get access to the pom from here.