Mule, Esper CEP and Non-Events

I was reading John D’Emic’s brilliant post I was amazed how good they suit each other. Letting Mule do the hard work of shuffling data and intelligent routing while Esper analyzing events on a overall level. It gave me kind of a opportunity to be stateful in a stateless environment. For those who aren’t familiar with Esper, it is what is called a Complex Event Processing engine that allows you to create queries and assertions on streams of events in real-time.

So the first thing that came to my mind was to address an issue that always bothered me in different solutions, to react on non-events.

Let’s say you have an integration that triggers on a file arriving at a certain place from an external source. So if that’s not happening no errors will be raised and it might take way too long time before it’s discovered. In Mule that’s kind of hard to handle since you have to know when the last time a file was processed. Instead you could make Mule tell Esper when a file is processed and let Esper decide when it’s time to alert on missing files.

I’ve put together a simple example on how this could be done.

First off we create the Esper config to define what kind of event that will be in play.

<?xml version="1.0" encoding="UTF-8"?>
<esper-configuration xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
                     xmlns="http://www.espertech.com/schema/esper"
                     xsi:schemaLocation="http://www.espertech.com/schema/esper http://www.espertech.com/schema/esper/esper-configuration-2.0.xsd">

    <event-type name="MyFileEvent" class="java.io.File"/>
    
</esper-configuration>

And then the Mule config

<?xml version="1.0" encoding="UTF-8"?>
<mule xmlns="http://www.mulesoft.org/schema/mule/core"
      xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
      xmlns:file="http://www.mulesoft.org/schema/mule/file"
      xmlns:spring="http://www.springframework.org/schema/beans"
      xmlns:script="http://www.mulesoft.org/schema/mule/scripting"
      xmlns:esper="http://www.mulesoft.org/schema/mule/esper"
      xsi:schemaLocation="
          http://www.mulesoft.org/schema/mule/core http://www.mulesoft.org/schema/mule/core/3.2/mule.xsd
          http://www.mulesoft.org/schema/mule/file http://www.mulesoft.org/schema/mule/file/3.2/mule-file.xsd
          http://www.mulesoft.org/schema/mule/scripting http://www.mulesoft.org/schema/mule/scripting/3.2/mule-scripting.xsd 
    	  http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-3.0.xsd
          http://www.mulesoft.org/schema/mule/esper http://www.mulesoft.org/schema/mule/esper/1.0/mule-esper.xsd
      ">
    	
    <file:connector name="input" pollingFrequency="100" 
					moveToDirectory="/tmp/mule-esper/processed" 
					moveToPattern="#[header:originalFilename]" 
					streaming="false"
					autoDelete="false">
		<service-overrides messageFactory="org.mule.transport.file.FileMuleMessageFactory" />
	</file:connector>					
	
	<esper:config configuration="esper.xml"/>
	
	<flow name="fileMoveFlow">
		<file:inbound-endpoint connector-ref="input" path="/tmp/mule-esper/inbox"  />
		<wire-tap>
			<processor-chain>
				<logger category="fileMoveFlow" level="INFO" message="File in process..."/>
				<esper:send eventPayload-ref="#[payload]" eventName="MyFileEvent" />
			</processor-chain>
		</wire-tap>
		<file:outbound-endpoint path="/tmp/mule-esper/outbox" />
	</flow>
	
	<flow name="EsperEventListenerFlow">
        <esper:listen statement="select * from pattern [every (timer:interval(30 sec) and not MyFileEvent)]"/> 
        <logger category="EsperEventListenerFlow" level="WARN" message="No files received the last 30 secs"/>
    </flow>
    
</mule>

The Esper listen statement claims that if no files arrives in 30 secs a warning should be logged. This of course should result in a mail being sent or whatever suits the overall monitoring system or support organization.  A info log entry is made whenever  files arrive. So looking at the log it shows that as long as files keep coming, it’s all happy days. But when there’s no file arriving in 30 secs a warning is raised.

15:09:53,325  INFO fileMoveFlow,[nonevents-0.0.1-SNAPSHOT].fileMoveFlow.stage1.02:197 - File in process...
15:09:59,455  INFO fileMoveFlow,[nonevents-0.0.1-SNAPSHOT].fileMoveFlow.stage1.02:197 - File in process...
15:10:13,367  INFO fileMoveFlow,[nonevents-0.0.1-SNAPSHOT].fileMoveFlow.stage1.02:197 - File in process...
15:10:21,710  INFO fileMoveFlow,[nonevents-0.0.1-SNAPSHOT].fileMoveFlow.stage1.02:197 - File in process...
15:10:51,719  WARN EsperEventListenerFlow,[nonevents-0.0.1-SNAPSHOT].EsperEventListenerFlow.stage1.02:269 - No files received the last 30 secs

Easy done and with a small amount of config.

This could easily be extended to report statistics. For example, to see how many files that arrived the last minute just add a flow with a new listener, like:

<flow name="EsperCountEventListenerFlow">
   <esper:listen statement="select count(*) from MyFileEvent.win:time(1 min)"/> 
   <logger category="EsperEventListenerFlow" level="INFO" message="Number of files last minute: #[groovy:payload.get('count(*)')]"/>
</flow>

Done! Once again, I really like the combination of Mule and Esper and how easy it is to get started.

This is a guest post from Mule community member Tomas Blohm. Thank you Tom!  If anyone else in the Mule community would like to write a guest post, please email us and get a cool T-shirt, squeezy Mule or other swag.

Related posts:

1. Twitter Complex Event Processing (CEP) with Esper and Drools
2. Mule Tip: Multiple PropertyPlaceholders in same Mule Configuration