@Test public void testProcessStreamingEvent() throws RuntimeAggregatorException, RainbowException, EventProcessingException { System.out.println("processStreamingEvent"); AcmeRuntimeAggregator instance = new AcmeRuntimeAggregator(systemConfig, new DummyDatabaseConnector()); instance.start(); IEventBuffer buffer = instance.getEventBuffer(); assertNotNull(buffer); buffer.activate(); assertEquals(true, buffer.isActive()); IRainbowMessage event = new RainbowESEBMessage(); event.setProperty(ESEBConstants.MSG_CHANNEL_KEY, "MODEL_DS"); event.setProperty(ESEBConstants.MSG_SENT, System.currentTimeMillis() + ""); instance.processEvent("MODEL_DS", event); List<IRainbowMessage> bufList = new ArrayList<>(); int num = buffer.drainToCollection(bufList); assertEquals(1, num); IRainbowMessage result = bufList.get(0); assertEquals(event.toString(), result.toString()); }
protected void processStreamingEvent(String channel, IRainbowMessage event) { eventBuffer.add(event); databaseConnector.writeEvent(channel, event, getEventTimestamp(event)); }
AcmeRuntimeAggregator implements IRuntimeAggregator<IAcmeSystem> { protected void processStreamingEvent(String channel, IRainbowMessage event) { eventBuffer.add(event); databaseConnector.writeEvent(channel, event, getEventTimestamp(event)); } }
AcmeRuntimeAggregator implements IRuntimeAggregator<IAcmeSystem> { protected void processStreamingEvent(String channel, IRainbowMessage event) { eventBuffer.add(event); databaseConnector.writeEvent(channel, event, getEventTimestamp(event)); } AcmeRuntimeAggregator(ISystemConfiguration config);  AcmeRuntimeAggregator(ISystemConfiguration config,
            IDatabaseConnector dbConn); }
AcmeRuntimeAggregator implements IRuntimeAggregator<IAcmeSystem> { protected void processStreamingEvent(String channel, IRainbowMessage event) { eventBuffer.add(event); databaseConnector.writeEvent(channel, event, getEventTimestamp(event)); } 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> { protected void processStreamingEvent(String channel, IRainbowMessage event) { eventBuffer.add(event); databaseConnector.writeEvent(channel, event, getEventTimestamp(event)); } 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);  }