Unit Testing Esper Statements

Posted on Thursday, May 13, 2010


Esper Statements would allow the flow of events to pass through them. Usually what would you like to unit test for a basic scenario?

  1. The valid events are being passed to the listener
  2. The invalid events are filtered out and are not passed to the listener
  3. The batching time, if defined for the statement, is being adhered to.

Let us quickly look at what a simple unit test would look like to cover all the above testing needs.

This is what our statement looks like

public void createStatement(EPAdministrator admin) {
		statement = admin.createEPL("insert into CompanyNewsOutputAlert(companyId, elementType, news) "
				+ "select InputEvent.companyId as companyId, InputEvent.elementType as elementType, InputEvent.news as news "
				+ "from CompanyNewsInputEvent.win:time_batch(5 sec) as InputEvent "
				+ "where InputEvent.elementType=com.inphina.cep.event.NewsElementType.NEWS");

The main things to notice in this statement are that we are interested in only CompanyNewsInputEvent and that the event stream of valid events should be batched for a period of 5 sec. Ok, let us see what the test class looks like

public class CompanyNewsStatementTest {

	private EPServiceProvider epService;
	private ApplicationContext ctx;
	private CompanyNewsStatement stmt;
	private FakeCompanyNewsEventListener listener;

	public void setUp() {

		// Get Beans from Spring container
		ctx = new ClassPathXmlApplicationContext("testApplicationContext.xml");
		stmt = (CompanyNewsStatement) ctx.getBean("companyNewsStatement");
		listener = (FakeCompanyNewsEventListener) stmt.getCompanyNewsEventListener();


		// Use external clocking for the test
		epService.getEPRuntime().sendEvent(new TimerControlEvent(TimerControlEvent.ClockType.CLOCK_EXTERNAL));

	private void startEsperEngine() {
		Configuration configuration = new Configuration();
		configuration.addEventType("CompanyNewsInputEvent", CompanyNewsInputEvent.class.getName());
		configuration.addEventType("CompanyScoreInputEvent", CompanyScoreInputEvent.class.getName());
		epService = EPServiceProviderManager.getProvider("CompanyNewsStatementTest", configuration);


As you would notice that in this test, we are injecting a fake listener. Since we are unit testing, we would be fine if we can test the statement. We do not want to test the listener logic. We use Spring to inject the FakeCompanyNewsEventListener into the Statement class. Our fake listener has utility methods to let us know if it is invoked and if it is then how many events does it have.

public class FakeCompanyNewsEventListener implements UpdateListener {

	private boolean isInvoked;
	private EventBean[] newData = {};

	public void update(EventBean[] newData, EventBean[] oldData) {
		this.newData = newData;

		isInvoked = true;

	public boolean isInvoked() {
		return isInvoked;

	public int getSizeOfEvents() {
		return newData.length;


So we are all set, now let us look at the test

	public void flowStreamOfEvents() throws InterruptedException {
		sendEvent(new CurrentTimeEvent(1000));

		sendEvent(new CompanyNewsInputEvent(100, NewsElementType.NEWS, "bla"));
		sendEvent(new CompanyNewsInputEvent(101, NewsElementType.NEWS, "bla"));
		sendEvent(new CompanyNewsInputEvent(102, NewsElementType.NEWS, "bla"));
		sendEvent(new CompanyNewsInputEvent(103, NewsElementType.NEWS, "bla"));
		sendEvent(new CompanyNewsInputEvent(104, NewsElementType.NEWS, "bla"));
		sendEvent(new CompanySMSInputEvent(104, SMSType.SHORT, "bla"));

		sendEvent(new CurrentTimeEvent(5000));
		Assert.assertEquals(0, listener.getSizeOfEvents());

		sendEvent(new CurrentTimeEvent(6000));

		Assert.assertEquals(5, listener.getSizeOfEvents());

	private void sendEvent(Object event) {

Here, amongst the news events, we are also sending the SMS event. Obviously our statement should not send that event to the listener and hence we have asserted for 5.
Also, if you have noticed the setup, you would notice another interesting thing. We are telling the Esper engine to use and external clock for clocking.

// Use external clocking for the test
		epService.getEPRuntime().sendEvent(new TimerControlEvent(TimerControlEvent.ClockType.CLOCK_EXTERNAL));

Using this external clock ensures that we can tell the Esper engine how much time has elapsed. So instead of waiting for 5 Secs to let the batching happen, we can tell the Esper engine that the time now is 6s, using a statement like this

sendEvent(new CurrentTimeEvent(6000));

Imagine what would do without this if your batch period was 24 hours. Of course you could also inject the batch period from your spring context as a property. If that is not the case then this is pretty handy.

Posted in: Java