EsperOperator implements DelayedResponseOperator { public void initialize(Properties properties) throws RequiredInputMissingException, ComponentInitializationFailedException { if(properties == null) throw new RequiredInputMissingException("Missing required properties"); Set<String> esperQueryStrings = new HashSet<>(); for(int i = 1; i < Integer.MAX_VALUE; i++) { String tmpStr = properties.getProperty(CFG_ESPER_STATEMENT_PREFIX+i); if(StringUtils.isBlank(tmpStr)) break; esperQueryStrings.add(StringUtils.trim(tmpStr)); } if(esperQueryStrings.isEmpty()) throw new RequiredInputMissingException("Missing required ESPER statement(s)"); Map<String, Map<String, String>> eventConfiguration = new HashMap<>(); for(int i = 1; i < Integer.MAX_VALUE; i++) { final String typeDefEvent = StringUtils.trim(properties.getProperty(CFG_ESPER_TYPE_DEF_PREFIX + i + CFG_ESPER_TYPE_DEF_EVENT_SUFFIX)); if(StringUtils.isBlank(typeDefEvent)) break; final String typeDefName = StringUtils.trim(properties.getProperty(CFG_ESPER_TYPE_DEF_PREFIX + i + CFG_ESPER_TYPE_DEF_NAME_SUFFIX)); final String typeDefType = StringUtils.trim(properties.getProperty(CFG_ESPER_TYPE_DEF_PREFIX + i + CFG_ESPER_TYPE_DEF_TYPE_SUFFIX)); if(StringUtils.isBlank(typeDefName) || StringUtils.isBlank(typeDefType)) throw new RequiredInputMissingException("Missing type def name or type for event '"+typeDefEvent+"' at position " + i); Map<String, String> ec = eventConfiguration.get(typeDefEvent); if(ec == null) ec = new HashMap<>(); ec.put(typeDefName, typeDefType); eventConfiguration.put(typeDefEvent, ec); } Configuration esperConfiguration = new Configuration(); for(final String event : eventConfiguration.keySet()) { Map<String, String> ec = eventConfiguration.get(event); if(ec != null && !ec.isEmpty()) { Map<String, Object> typeDefinition = new HashMap<>(); for(final String typeDefName : ec.keySet()) { final String typeDefType = ec.get(typeDefName); try { typeDefinition.put(typeDefName, Class.forName(typeDefType)); } catch(ClassNotFoundException e) { throw new ComponentInitializationFailedException("Failed to lookup provided type '"+typeDefType+"' for event '"+event+"'. Error: " + e.getMessage()); } } esperConfiguration.addEventType(event, typeDefinition); } } Map<String, Object> spqrDefaultTypeDefinition = new HashMap<>(); spqrDefaultTypeDefinition.put(SPQR_EVENT_TIMESTAMP_FIELD, Long.class); spqrDefaultTypeDefinition.put(SPQR_EVENT_BODY_FIELD, Map.class); esperConfiguration.addEventType(DEFAULT_INPUT_EVENT, spqrDefaultTypeDefinition); esperConfiguration.addEventType(DEFAULT_OUTPUT_EVENT, spqrDefaultTypeDefinition); this.esperServiceProvider = EPServiceProviderManager.getDefaultProvider(esperConfiguration); this.esperServiceProvider.initialize(); for(final String qs : esperQueryStrings) { try { EPStatement esperStatement = this.esperServiceProvider.getEPAdministrator().createEPL(qs); esperStatement.setSubscriber(this); } catch(EPStatementException e) { throw new ComponentInitializationFailedException("Failed to parse query into ESPER statement. Error: " + e.getMessage(), e); } } this.esperRuntime = this.esperServiceProvider.getEPRuntime(); }  void initialize(Properties properties); boolean shutdown(); MicroPipelineComponentType getType(); void onMessage(StreamingDataMessage message); StreamingDataMessage[] getResult(); void update(Map<String, Object> eventMap); long getNumberOfMessagesSinceLastResult(); long getTotalNumOfMessages(); void setId(String id); String getId(); void setWaitStrategy(DelayedResponseOperatorWaitStrategy waitStrategy); static final String SPQR_EVENT_TIMESTAMP_FIELD; static final String SPQR_EVENT_BODY_FIELD; static final String DEFAULT_INPUT_EVENT; static final String DEFAULT_OUTPUT_EVENT; static final String CFG_ESPER_STATEMENT_PREFIX; static final String CFG_ESPER_TYPE_DEF_PREFIX; static final String CFG_ESPER_TYPE_DEF_EVENT_SUFFIX; static final String CFG_ESPER_TYPE_DEF_NAME_SUFFIX; static final String CFG_ESPER_TYPE_DEF_TYPE_SUFFIX; }
@Test public void testInitialize_withNullInput() throws ComponentInitializationFailedException { try { new EsperOperator().initialize(null); Assert.fail("Invalid input"); } catch(RequiredInputMissingException e) { } }
@Test public void testInitialize_withPropertiesMissingFieldType() throws ComponentInitializationFailedException { Properties properties = new Properties(); properties.put(EsperOperator.CFG_ESPER_TYPE_DEF_PREFIX+"1"+EsperOperator.CFG_ESPER_TYPE_DEF_EVENT_SUFFIX, "testEvent"); properties.put(EsperOperator.CFG_ESPER_TYPE_DEF_PREFIX+"1"+EsperOperator.CFG_ESPER_TYPE_DEF_NAME_SUFFIX, "fieldName"); properties.put(EsperOperator.CFG_ESPER_STATEMENT_PREFIX+"1", "select * from testEvent"); try { new EsperOperator().initialize(properties); Assert.fail("Invalid input"); } catch(RequiredInputMissingException e) { } }
@Test public void testInitialize_withPropertiesMissingFieldName() throws ComponentInitializationFailedException { Properties properties = new Properties(); properties.put(EsperOperator.CFG_ESPER_TYPE_DEF_PREFIX+"1"+EsperOperator.CFG_ESPER_TYPE_DEF_EVENT_SUFFIX, "testEvent"); properties.put(EsperOperator.CFG_ESPER_TYPE_DEF_PREFIX+"1"+EsperOperator.CFG_ESPER_TYPE_DEF_TYPE_SUFFIX, "java.lang.String"); properties.put(EsperOperator.CFG_ESPER_STATEMENT_PREFIX+"1", "select * from testEvent"); try { new EsperOperator().initialize(properties); Assert.fail("Invalid input"); } catch(RequiredInputMissingException e) { } }
@Test public void testInitialize_withPropertiesInvalidType() throws RequiredInputMissingException { Properties properties = new Properties(); properties.put(EsperOperator.CFG_ESPER_TYPE_DEF_PREFIX+"1"+EsperOperator.CFG_ESPER_TYPE_DEF_EVENT_SUFFIX, "testEvent"); properties.put(EsperOperator.CFG_ESPER_TYPE_DEF_PREFIX+"1"+EsperOperator.CFG_ESPER_TYPE_DEF_NAME_SUFFIX, "fieldName"); properties.put(EsperOperator.CFG_ESPER_TYPE_DEF_PREFIX+"1"+EsperOperator.CFG_ESPER_TYPE_DEF_TYPE_SUFFIX, "java.lang.NoSuchClass"); properties.put(EsperOperator.CFG_ESPER_STATEMENT_PREFIX+"1", "select * from testEvent"); try { new EsperOperator().initialize(properties); Assert.fail("Invalid input"); } catch(ComponentInitializationFailedException e) { } }
@Test public void testInitialize_withStatementOnly() throws ComponentInitializationFailedException, RequiredInputMissingException { Properties properties = new Properties(); properties.put(EsperOperator.CFG_ESPER_STATEMENT_PREFIX+"1", "select * from spqrIn"); new EsperOperator().initialize(properties); }
@Test public void testInitialize_withPropertiesWithoutStatement() throws ComponentInitializationFailedException { Properties properties = new Properties(); properties.put(EsperOperator.CFG_ESPER_TYPE_DEF_PREFIX+"1"+EsperOperator.CFG_ESPER_TYPE_DEF_EVENT_SUFFIX, "testEvent"); properties.put(EsperOperator.CFG_ESPER_TYPE_DEF_PREFIX+"1"+EsperOperator.CFG_ESPER_TYPE_DEF_NAME_SUFFIX, "fieldName"); properties.put(EsperOperator.CFG_ESPER_TYPE_DEF_PREFIX+"1"+EsperOperator.CFG_ESPER_TYPE_DEF_TYPE_SUFFIX, "java.lang.String"); try { new EsperOperator().initialize(properties); Assert.fail("Invalid input"); } catch(RequiredInputMissingException e) { } }
@Test public void testInitialize_withPropertiesInvalidStatement() throws RequiredInputMissingException { Properties properties = new Properties(); properties.put(EsperOperator.CFG_ESPER_TYPE_DEF_PREFIX+"1"+EsperOperator.CFG_ESPER_TYPE_DEF_EVENT_SUFFIX, "testEvent"); properties.put(EsperOperator.CFG_ESPER_TYPE_DEF_PREFIX+"1"+EsperOperator.CFG_ESPER_TYPE_DEF_NAME_SUFFIX, "fieldName"); properties.put(EsperOperator.CFG_ESPER_TYPE_DEF_PREFIX+"1"+EsperOperator.CFG_ESPER_TYPE_DEF_TYPE_SUFFIX, "java.lang.String"); properties.put(EsperOperator.CFG_ESPER_STATEMENT_PREFIX+"1", "select * from noSuchEventType"); try { new EsperOperator().initialize(properties); Assert.fail("Invalid input"); } catch(ComponentInitializationFailedException e) { } }