@Test public void testProcessEvent() throws EventProcessingException, RainbowException, RuntimeAggregatorException { System.out.println("processEvent"); AcmeRuntimeAggregator instance = new AcmeRuntimeAggregator(systemConfig, new DummyDatabaseConnector()); String channel = "MODEL_CHANGE"; Float expRes = (float) 99.9; IRainbowMessage event = new RainbowESEBMessage(); event.setProperty(IRainbowMessageFactory.EVENT_TYPE_PROP, AcmeModelEventType.SET_PROPERTY_VALUE.toString()); event.setProperty(AcmeModelOperation.PROPERTY_PROP, "ZNewsSys.Server0.load"); event.setProperty(AcmeModelOperation.VALUE_PROP, expRes.toString()); event.setProperty(ESEBConstants.MSG_SENT, System.currentTimeMillis() + ""); instance.start(); instance.processEvent(channel, event); IModelInstance<IAcmeSystem> modelInstance = instance.getInternalModel(); assertNotNull(modelInstance); IAcmeSystem system = modelInstance.getModelInstance(); assertNotNull(system); IAcmeProperty prop = (IAcmeProperty) system.getComponent("Server0").getProperty("load"); assertNotNull(prop); IAcmePropertyValue val = prop.getValue(); assertNotNull(val); Float res = PropertyHelper.toJavaVal((IAcmeFloatValue) val); assertEquals(expRes, res); try { instance.processEvent("UNKNOWN", event); } catch (EventProcessingException ex) { } instance.stop(); }
@Override synchronized public void processEvent(String channel, IRainbowMessage event) throws EventProcessingException { if (!isRunning) { return; } switch (channel) { case "MODEL_CHANGE": processModelUpdate(event); break; case "MODEL_DS": case "MODEL_US": processStreamingEvent(channel, event); break; default: throw new EventProcessingException("Unsupported channel: " + channel); } }
AcmeRuntimeAggregator implements IRuntimeAggregator<IAcmeSystem> { @Override synchronized public void processEvent(String channel, IRainbowMessage event) throws EventProcessingException { if (!isRunning) { return; } switch (channel) { case "MODEL_CHANGE": processModelUpdate(event); break; case "MODEL_DS": case "MODEL_US": processStreamingEvent(channel, event); break; default: throw new EventProcessingException("Unsupported channel: " + channel); } } }
AcmeRuntimeAggregator implements IRuntimeAggregator<IAcmeSystem> { @Override synchronized public void processEvent(String channel, IRainbowMessage event) throws EventProcessingException { if (!isRunning) { return; } switch (channel) { case "MODEL_CHANGE": processModelUpdate(event); break; case "MODEL_DS": case "MODEL_US": processStreamingEvent(channel, event); break; default: throw new EventProcessingException("Unsupported channel: " + channel); } } AcmeRuntimeAggregator(ISystemConfiguration config);  AcmeRuntimeAggregator(ISystemConfiguration config,
            IDatabaseConnector dbConn); }
AcmeRuntimeAggregator implements IRuntimeAggregator<IAcmeSystem> { @Override synchronized public void processEvent(String channel, IRainbowMessage event) throws EventProcessingException { if (!isRunning) { return; } switch (channel) { case "MODEL_CHANGE": processModelUpdate(event); break; case "MODEL_DS": case "MODEL_US": processStreamingEvent(channel, event); break; default: throw new EventProcessingException("Unsupported channel: " + channel); } } AcmeRuntimeAggregator(ISystemConfiguration config);  AcmeRuntimeAggregator(ISystemConfiguration config,
            IDatabaseConnector dbConn); @Override void start(); @Override void stop(); @Override IModelInstance<IAcmeSystem> getInternalModel(); @Override IModelInstance<IAcmeSystem> copyInternalModel(); @Override IEventBuffer getEventBuffer(); @Override synchronized void processEvent(String channel, IRainbowMessage event); }
AcmeRuntimeAggregator implements IRuntimeAggregator<IAcmeSystem> { @Override synchronized public void processEvent(String channel, IRainbowMessage event) throws EventProcessingException { if (!isRunning) { return; } switch (channel) { case "MODEL_CHANGE": processModelUpdate(event); break; case "MODEL_DS": case "MODEL_US": processStreamingEvent(channel, event); break; default: throw new EventProcessingException("Unsupported channel: " + channel); } } AcmeRuntimeAggregator(ISystemConfiguration config);  AcmeRuntimeAggregator(ISystemConfiguration config,
            IDatabaseConnector dbConn); @Override void start(); @Override void stop(); @Override IModelInstance<IAcmeSystem> getInternalModel(); @Override IModelInstance<IAcmeSystem> copyInternalModel(); @Override IEventBuffer getEventBuffer(); @Override synchronized void processEvent(String channel, IRainbowMessage event);  }