Howto embed & configure Mule 2.0 and ActiveMQ in a Spring web application

Embedded Mule, ActiveMQ & Spring

Overview 

This post illustrates how to setup an open source Enterprise Service Bus (ESB), messaging broker and JMS 1.1 provider using Spring integration.  

After spending some time wiring up Mule 1.4.3 and ActiveMQ into a webapp with Spring, and ultimately deciding to make the jump to Mule 2.0 M2 (very close to a 2.0 final) I decided to post the steps I took, my lessons learned and my end Spring configuration snippets.  While there may soon be plenty of documentation on how to do this (once Mule 2.0 is released) this guide has some early adopter documentation that is not currently widely available.

Downloads

What you'll need to add to your WEB-INF/lib (since we're adding this to a new or existing web application):

• Spring 2.0 (2.0 because both ActiveMQ and Mule leverage Spring 2.0 namespace handlers) - spring-2.x.x.jar
• Mule 2.0 (currently, just a technology preview is available) - mule-core-2.0.0-M2.jar plus the module-builders, module-spring, transport-email, transport-jms, transport-stdio, transport-tcp, transport-vm JARs.
  
• ActiveMQ 4.1.1 ( open source Message Broker with full JMS 1.1 support) - activemq-core-4.1.1.jar
• Other dependencies as needed (grab these from apache-activemq-4.1.1/lib/optional and mule-2.0.0-M2/lib/opt from your unpacked mule and activemq downloads from above
• MySQL
  

Configuring ActiveMQ

 ActiveMQ can be embedded into your web application or can be run stand alone.  I'll show you how to do run the broker in both ways.  First, we'll look at the stand alone approach:

• linux-command: cd apache-activemq-4.1.1/conf
• linux-command: cp activemq.xml my-activemq.xml
• edit my-activemq.xml:
  
• uncomment the <managementContext> block so we can monitor the JMS server via JMX
• in the <persistenceAdapter> block, comment out the existing persistence adapters and add this one:
  
• <jdbcPersistenceAdapter useDatabaseLock="false" dataSource="#mysql-ds"/>
• uncomment the <bean id="mysql-ds" ... > </bean> block
  
• gotcha: if you have MySQL installed and your my.ini (or my.cnf) is configured to use default-character-set=utf8 (and possibly only if you are using a version of MySQL earlier than 5.0), then you'll need to manually create MyISAM SQL tables used by ActiveMQ to store JMS messages.  The reason is that a single-byte character set instead of a Unicode variant, causes the key columns to exceed the maximum index length.
  
• command:  mysql -uroot -p
• mysql> create database activemq;
• mysql> grant all privileges on activemq.* to 'activemq'@'localhost' identified by 'activemq' with grant option;
• mysql> exit;
• command: mysql -uactivemq -pactivemq activemq < activemq.sql
• (please post a comment and ask to be sent this activemq.sql) 

Starting stand alone

• linux-command:  cd apache-activemq-4.1.1
  
• linux-command:  ./bin/activemq xbean:../conf/my-activemq.xml

That's it!  ActiveMQ is currently running and you can monitor it via jconsole (a JMX tool that ships with JDK 1.5 - you can start it by typing 'jconsole' on a command line that has $JAVA_HOME/bin in the path) and connecting locally or via this JMX URL: service:jmx:rmi:///jndi/rmi://localhost:1099/jmxrmi

Embedding into your web application

As I'm assuming that you already have a Spring web application, so simply adding these beans will get ActiveMQ started with your web application.  (Note the use of the ActiveMQ namespace.  I also have xbean-spring-2.8.jar in my classpath.)

<beans
  xmlns="http://www.springframework.org/schema/beans"
  xmlns:amq="http://activemq.org/config/1.0"
  xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
  xsi:schemaLocation=
  "http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-2.0.xsd
  http://activemq.org/config/1.0 http://activemq.apache.org/snapshot-schema/activemq-core-5.0-SNAPSHOT.xsd">
 
  <!--  lets create an embedded ActiveMQ Broker -->
  <amq:broker brokerName="brokerFactory" useJmx="true" persistent="false">
    <amq:managementContext >
        <amq:managementContext connectorPort="1099" jmxDomainName="org.apache.activemq"/>
    </amq:managementContext>
    <amq:persistenceAdapter>
        <amq:jdbcPersistenceAdapter useDatabaseLock="false" dataSource="#mysql-ds"/>
    </amq:persistenceAdapter>
    <amq:transportConnectors>
      <amq:transportConnector name="openwire" uri="tcp://localhost:61616" discoveryUri="multicast://default" />
      <amq:transportConnector name="ssl" uri="ssl://localhost:61617"  />
    </amq:transportConnectors>
  </amq:broker>

  <bean id="mysql-ds" class="org.apache.commons.dbcp.BasicDataSource" destroy-method="close">
    <property name="driverClassName" value="com.mysql.jdbc.Driver"/>
    <property name="url" value="jdbc:mysql://localhost/activemq?relaxAutoCommit=true"/>
    <property name="username" value="activemq"/>
    <property name="password" value="activemq"/>
    <property name="poolPreparedStatements" value="true"/>
  </bean>
 
</beans>

 

gotcha: Starting ActiveMQ as a stand alone server (on separate hardware or VM) obviously has strong advantages, but for the purposes of a personal challenge of trying to wire these components (Mule and ActiveMQ) up as embedded components started by Spring, I was disappointed to find that my CPU was pegged at 100% whenever I embedded ActiveMQ in this way.  Apparently this is a known issue and I'm trying to figure out when it will be fixed.  This high CPU issue may also be a problem with the Java Service Wrapper, which is used by both ActiveMQ and Mule .  There have been reports of this wrapper behaving differently between Windows and Linux platforms.

Creating a Spring JmsTemplate and Queues

I've placed the following configuration in applicationContext-jms.xml, which is loaded into the Spring context.  This config accomplishes the following things:

• Creates a connection pool for ActiveMQ connections>
• Creates a Spring JMSTemplate> in order to simplify synchronous JMS access code
• Creates three JMS queues (that we'll later reference from within the Mule configuration)
• Declares a MailsenderProducer class that  places a message on a queue (so we can test that Mule  properly monitors this queue and processes messages on the queue)

    <bean id="jmsPooledConnectionFactory" name="jmsPooledConnectionFactory"
        class="org.apache.activemq.spring.ActiveMQConnectionFactory">
        <property name="brokerURL" value="vm://localhost" />
    </bean>   
   
    <bean id="myJmsTemplate" class="org.springframework.jms.core.JmsTemplate">
      <property name="connectionFactory">
        <ref local="jmsPooledConnectionFactory"/>
      </property>
    </bean>       
       
    <bean id="destination" class="org.apache.activemq.command.ActiveMQQueue" autowire="constructor">
        <constructor-arg>
            <value>sai.test.queue</value>
        </constructor-arg>
    </bean>
   
    <bean id="emailToProcessQueue" class="org.apache.activemq.command.ActiveMQQueue" autowire="constructor">
        <constructor-arg>
            <value>sai.unprocessed.email.queue</value>
        </constructor-arg>
    </bean>
   
    <bean id="processedEmailNotificationQueue" class="org.apache.activemq.command.ActiveMQQueue" autowire="constructor">
        <constructor-arg>
            <value>sai.processed.email.queue</value>
        </constructor-arg>
    </bean>
   

    <bean id="producer" class="com.sourceallies.proj.mq.MailsenderProducer">
        <property name="template">
            <ref bean="myJmsTemplate"/>
        </property>
        <property name="destination">
            <ref bean="destination"/>
        </property>
    </bean>

Configuring Mule

Adding a servlet listener

You need to add the following elements to your web.xml deployment descriptor in order to start Mule in your web application.  Mule, like ActiveMQ, can be started  stand alone or from within Spring.   This post  only shows how to start Mule from within Spring.

    <context-param>
        <param-name>org.mule.webapp.classpath</param-name>
        <param-value>/WEB-INF/classes</param-value>
    </context-param>

    <context-param>
        <param-name>org.mule.config</param-name>
        <param-value>
            applicationContext-mule.xml
        </param-value>
    </context-param>

    <listener>
        <listener-class>org.mule.config.builders.MuleXmlBuilderContextListener</listener-class>
    </listener>>

 

 gotcha:  Simply including the Mule beans in your Spring context will not start the Mule server.  This listener is explicitly needed to start Mule.  Mule 1.4.x users: You no longer need to include the AutowireUMOManagerFactoryBean, (renamed to ManagementContextFactoryBean in Mule 2.0) in your spring config.  This bean, and others, are included by default in the default-mule-config.xml file (tucked away inside of the mule-core-2.x.x.jar).

Wiring the Beans

Below is the header of a sample Mule configuration file.  Notice how Mule has the default namespace and Spring beans need to be qualified by spring:  This is the file that is referenced in the org.mule.conf web.xml context parameter.  It does not appear to matter whether or not this file is also added to the Spring contextConfigLocation web.xml context parameter.

 <?xml version="1.0" encoding="UTF-8"?>
<mule xmlns="http://www.mulesource.org/schema/mule/core/2.0"
       xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
       xmlns:spring="http://www.springframework.org/schema/beans"
       xmlns:stdio="http://www.mulesource.org/schema/mule/stdio/2.0"
       xmlns:jms="http://www.mulesource.org/schema/mule/jms/2.0"
       xmlns:pop3="http://www.mulesource.org/schema/mule/pop3/2.0"
       xmlns:smtp="http://www.mulesource.org/schema/mule/smtp/2.0"
       xmlns:vm="http://www.mulesource.org/schema/mule/vm/2.0"
       xmlns:email-download="http://www.mulesource.org/schema/mule/email-download/2.0"
    xsi:schemaLocation="
       http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-2.0.xsd
       http://www.mulesource.org/schema/mule/core/2.0 http://www.mulesource.org/schema/mule/core/2.0/mule.xsd
       http://www.mulesource.org/schema/mule/stdio/2.0 http://www.mulesource.org/schema/mule/stdio/2.0/mule-stdio.xsd
       http://www.mulesource.org/schema/mule/jms/2.0 http://www.mulesource.org/schema/mule/jms/2.0/mule-jms.xsd
       http://www.mulesource.org/schema/mule/pop3/2.0 http://www.mulesource.org/schema/mule/pop3/2.0/mule-pop3.xsd
       http://www.mulesource.org/schema/mule/smtp/2.0 http://www.mulesource.org/schema/mule/smtp/2.0/mule-smtp.xsd
       http://www.mulesource.org/schema/mule/vm/2.0 http://www.mulesource.org/schema/mule/vm/2.0/mule-vm.xsd
       http://www.mulesource.org/schema/mule/email-download/2.0 http://www.mulesource.org/schema/mule/email-download/2.0/mule-email-download.xsd">

 
Below is the rest of this Mule 2.0 configuration file.


    <!-- Synchronous VM connector (messages are processed in the same execution thread) -->
    <vm:connector name="default"/>
    <!-- Asynchronous VM connector (uses in-memory queues) -->
    <vm:connector name="queue" queueEvents="true"/>

    <spring:bean id="propertyPlaceholderConfigurer"
          class="org.springframework.beans.factory.config.PropertyPlaceholderConfigurer">
        <spring:property name="locations">
            <spring:list>
                <spring:value>email.properties</spring:value>
            </spring:list>
        </spring:property>
    </spring:bean>
   
    <pop3:connector name="pop3"/>

    <smtp:connector name="EmailConnector"
        fromAddress="sourceallies@localhost"
        username="sourceallies"
        password="sourceallies"
        subject="File Received!"
        contentType="text/plain" />

    <configuration serverId="vendorPortalEmbeddedMuleId">
        <default-threading-profile threadWaitTimeout="30" maxThreadsActive="2" maxThreadsIdle="2"
                                        maxBufferSize="0" threadTTL="60000" poolExhaustedAction="RUN"  />
    </configuration>

    <jms:activemq-connector name="jmsConnector" specification="1.1" >
        <dispatcher-threading-profile doThreading="true" maxThreadsActive="2" />
    </jms:activemq-connector>
  
     <custom-transformer name="ObjectToJMSMessage"
                                 class="org.mule.providers.jms.transformers.ObjectToJMSMessage" />
     <custom-transformer name="stringToJmsMessage"
                                 class="org.mule.providers.jms.transformers.ObjectToJMSMessage" >
         <spring:property name="returnClass"><spring:value>javax.jms.TextMessage</spring:value></spring:property>
     </custom-transformer>
     <custom-transformer name="jmsMessageToString"
                                 class="org.mule.providers.jms.transformers.JMSMessageToObject" >
                           <spring:property name="returnClass"><spring:value>java.lang.String</spring:value></spring:property>      
      </custom-transformer>
        
                                     
     <endpoint name="testEndpoint" address="jms://sai.test.queue" connector-ref="jmsConnector"/>
     <endpoint name="emailAccountEndpoint" address="pop3://sourceallies:sourceallies@localhost" />
     
     
     <endpoint name="emailToProcessEndpoint"
         address="jms://sai.unprocessed.email.queue" connector-ref="jmsConnector"/>         
    
     <endpoint name="processedEmailNotificationEndpoint"
         address="jms://sai.processed.email.queue" connector-ref="jmsConnector" />

    <!-- The Mule model initialises and manages your UMO components  -->
    
    <model name="saiEmbeddedMule" >
        <!--
            A Mule service defines all the necessary information about how your components will
            interact with the framework, other components in the system and external sources.
            Please refer to the Configuration Guide for a full description of all the parameters.
        -->
        
        <service name="echoUMO">
       
            <component class="org.mule.components.simple.EchoComponent"/>

            <!-- any number of endpoints can be added to an inbound router -->
            <inbound-router>
            
                <inbound-endpoint ref="testEndpoint" transformer-ref="jmsMessageToString"/>
                <inbound-endpoint ref="emailAccountEndpoint"></inbound-endpoint>
            </inbound-router>

            <!--
                An outbound router can have one or more router configurations that can be
                invoked depending on business rules, message contents, headers or any other
                criteria. The pass-through-router is a router that automatically passes
                on every message it receives
            -->
            <outbound-router>
                <outbound-pass-through-router>
                    <outbound-endpoint address="stdio://System.out"/>
                </outbound-pass-through-router>
            </outbound-router>
        </service>
        
        <service name="emailProcessorDescriptor">
              <component class="org.mule.components.simple.EchoComponent" />
             
              <inbound-router>
                  <inbound-endpoint ref="emailToProcessEndpoint"  />
              </inbound-router>

              <outbound-router>
                  <outbound-pass-through-router>
                      <outbound-endpoint ref="processedEmailNotificationEndpoint" />
                  </outbound-pass-through-router>
              </outbound-router>

         </service>      
    </model>    
</mule>

Gotcha:  Because of a bug in the pre-final Mule code, when monitoring a JMS queue there is not currently a way to set the monitoring interval (queueTimeout).  It is therefore set to zero, which leads to yet another high CPU usage problem.  Hopefully, by the time you read this, you have access to a more recent milestone release than M2, which should have this "100% CPU problem" fixed.

The Mule Beans Explained 

At this point (once we have all of the trouble of wiring up the ESB and JMS server), the Mule docs http://mule.mulesource.org/display/MULE/Programmers+Guide are a better and more accurate source for explaining what each of the beans do, but I'll do a quick run-through and include snippets of Mule docs where appropriate.

     <!-- Synchronous VM connector (messages are processed in the same execution thread) -->
    <vm:connector name="default"/>
    <!-- Asynchronous VM connector (uses in-memory queues) -->
    <vm:connector name="queue" queueEvents="true"/>

The config above provides an way to communicate with the Mule ESB within the virtual machine.

    <spring:bean id="propertyPlaceholderConfigurer"
          class="org.springframework.beans.factory.config.PropertyPlaceholderConfigurer">
        <spring:property name="locations">
            <spring:list>
                <spring:value>email.properties</spring:value>
            </spring:list>
        </spring:property>
    </spring:bean>
   
    <pop3:connector name="pop3"/>

    <smtp:connector name="EmailConnector"
        fromAddress="sourceallies@localhost"
        username="sourceallies"
        password="sourceallies"
        subject="File Received!"
        contentType="text/plain" />

This wires up Mule SMTP and POP3 connectors.  Connectors are used by Mule to register listeners and create message dispatchers for the transport of messages. Configuration parameters that should be shared by all Message Receivers and Dispatchers are stored on the connector.  Here, we're using mixing a Spring PropertyPlaceholderConfigurer (to externalize some POP3 config such as username & password) with a Mule connector that knows to look for these properties.

    <configuration serverId="vendorPortalEmbeddedMuleId">
        <default-threading-profile threadWaitTimeout="30" maxThreadsActive="2" maxThreadsIdle="2"
                                        maxBufferSize="0" threadTTL="60000" poolExhaustedAction="RUN"  />
    </configuration>

 The config above configures Mules thread usage.

    <jms:activemq-connector name="jmsConnector" specification="1.1" >
        <dispatcher-threading-profile doThreading="true" maxThreadsActive="2" />
    </jms:activemq-connector>  

The config above configures an ActiveMQ specific connector.  We reference the name "jmsConnector" in all other components/endpoints that require JMS connectivity.

     <custom-transformer name="ObjectToJMSMessage"
                                 class="org.mule.providers.jms.transformers.ObjectToJMSMessage" />
     <custom-transformer name="stringToJmsMessage"
                                 class="org.mule.providers.jms.transformers.ObjectToJMSMessage" >
         <spring:property name="returnClass"><spring:value>javax.jms.TextMessage</spring:value></spring:property>
     </custom-transformer>
     <custom-transformer name="jmsMessageToString"
                                 class="org.mule.providers.jms.transformers.JMSMessageToObject" >
                           <spring:property name="returnClass"><spring:value>java.lang.String</spring:value></spring:property>      
      </custom-transformer>
        

Transformers are used to convert inbound data to an object type required by your business object (UMO Component) or outbound to an object type required by the transport such as a JmsMessage as shown above.
                                     
     <endpoint name="testEndpoint" address="jms://sai.test.queue" connector-ref="jmsConnector"/>
     <endpoint name="emailAccountEndpoint" address="pop3://sourceallies:sourceallies@localhost" />
     <endpoint name="emailToProcessEndpoint"
         address="jms://sai.unprocessed.email.queue" connector-ref="jmsConnector"/>        
     <endpoint name="processedEmailNotificationEndpoint"
         address="jms://sai.processed.email.queue" connector-ref="jmsConnector" />

Endpoints are used to connect to components in the server and external systems or to each other locally or over the network. An Endpoint is a specific channel through which two parties can communicate.  Above, we've defined three endpoints to JMS queues and an endpoint to a pop3 e-mail account (running on a local James server).

    
    <model name="saiEmbeddedMule" >
        <service name="echoUMO">
            <component class="org.mule.components.simple.EchoComponent"/>
            <inbound-router>        
                <inbound-endpoint ref="testEndpoint" transformer-ref="jmsMessageToString"/>
                <inbound-endpoint ref="emailAccountEndpoint"></inbound-endpoint>
            </inbound-router>
            <outbound-router>
                <outbound-pass-through-router>
                    <outbound-endpoint address="stdio://System.out"/>
                </outbound-pass-through-router>
            </outbound-router>
        </service>
    </model>

Finally, we get to the purpose of all of this configuration: an example of Mule in action that does very little (as all good demos do).  We start by declaring the Mule model which initializes and manages our business objects that are fed messages from inbound-routers and that send messages on to outbound-routers.

 In this example, we use a Mule-provided component called EchoComponent that just takes input and repeats it as output.  EchoComponent is normally where you would place your business objects that process an incoming order for example.

 We listen on two different endpoints: an e-mail account and a JMS queue.  (Remember, this could be analogous to looking for incoming orders via an e-mail account that translates the e-mail text into an order object, or via the queue where a JMS TextMessage is converted into the same order object to be processed.)

 

Instead of actually processing the message with a business object, we use the EchoComponent and the outbound-pass-through-router (a router that automatically sends every message it receives) to send the message to and endpoint that simply drops the message to the console.

        <service name="emailProcessorDescriptor">
              <component class="org.mule.components.simple.EchoComponent" />
              <inbound-router>
                  <inbound-endpoint ref="emailToProcessEndpoint"  />
              </inbound-router>
              <outbound-router>
                  <outbound-pass-through-router>
                      <outbound-endpoint ref="processedEmailNotificationEndpoint" />
                  </outbound-pass-through-router>
              </outbound-router>
         </service>      

 

 

The mule config above is a another example that takes JMS messages from one queue and (instead of adding complexity to this example and processing them) simply drops them into another queue endpoint.  Because the endpoints we use reference the jmsConnector by name, both of these examples leverage the MySQL backed ActiveMQ configuration from the beginning of this post.

Hopefully this helps you get a jump start in using some of the soon to be released Mule 2.0 functionality.  Feel free to post questions or suggestions in the comments section below.

 

Posted at 03:27PM Oct 22, 2007 by mattvincent in Java  |  Comments[0]