Unit Testing Esper Statements

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

[sourcecode language=”java”]
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");
statement.addListener(companyNewsEventListener);
}
[/sourcecode]

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

[sourcecode language=”java”]
public class CompanyNewsStatementTest {

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

@Before
public void setUp() {

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

stmt.createStatement(epService.getEPAdministrator());

// 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);

epService.initialize();
}
[/sourcecode]

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.

[sourcecode language=”java”]
public class FakeCompanyNewsEventListener implements UpdateListener {

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

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

isInvoked = true;
}

public boolean isInvoked() {
return isInvoked;
}

public int getSizeOfEvents() {
return newData.length;
}

}
[/sourcecode]

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

[sourcecode language=”java”]
@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.assertFalse(listener.isInvoked());
Assert.assertEquals(0, listener.getSizeOfEvents());

sendEvent(new CurrentTimeEvent(6000));

Assert.assertTrue(listener.isInvoked());
Assert.assertEquals(5, listener.getSizeOfEvents());
}

private void sendEvent(Object event) {
epService.getEPRuntime().sendEvent(event);
}
[/sourcecode]

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.

[sourcecode language=”java”]
// Use external clocking for the test
epService.getEPRuntime().sendEvent(new TimerControlEvent(TimerControlEvent.ClockType.CLOCK_EXTERNAL));
[/sourcecode]

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.