SiddhiHandler implements RestListener, EventHandler<MapEvent> { @Override public void onEvent(MapEvent mapEvent, long sequence, boolean endOfBatch) throws Exception { if (!siddhiPlans.isEmpty()) { Map<String, Object> data = mapEvent.getData(); String topic = mapEvent.getSource(); for (SiddhiPlan siddhiPlan : siddhiPlans.values()) { if (siddhiPlan.getInput().contains(topic)) { siddhiPlan.send(topic, data); } } } } SiddhiHandler(Sink eventReceiver);  SiddhiHandler(); void stop(); Map<String, SiddhiPlan> getSiddhiPlans(); @Override void onEvent(MapEvent mapEvent, long sequence, boolean endOfBatch); @Override void add(Map<String, Object> element); synchronized void add(SiddhiPlan siddhiPlan); @Override synchronized void remove(String id); @Override synchronized void synchronize(List<Map<String, Object>> elements); @Override List<Map<String, Object>> list(); synchronized void save(); synchronized void restore();  }
@Test public void onEvent() throws Exception { SiddhiPlan siddhiPlanMock = mock(SiddhiPlan.class); when(siddhiPlanMock.getId()).thenReturn("testMockID"); when(siddhiPlanMock.getInput()).thenReturn(Collections.singletonList("test")); SiddhiHandler siddhiHandler = new SiddhiHandler(); siddhiHandler.add(siddhiPlanMock); Map<String, Object> dataMap = new HashMap<>(); dataMap.put("a", "testa"); dataMap.put("b", "testb"); dataMap.put("c", "testc"); dataMap.put("d", 128); dataMap.put("e", 10); MapEvent mapEventStub = new MapEvent(); mapEventStub.setData(dataMap); mapEventStub.setSource("test"); siddhiHandler.onEvent(mapEventStub, 1, true); verify(siddhiPlanMock).send("test", dataMap); }