## 5 Proposed Implementation for Parametric Event Transformation
### Abstraction and Reusable Implementation: Time Windowed Moving Average Calculation
Each time a new event enters the windowed stream, it is checked whether it should be used for averaging. The statement that reflects the business logic is the following: If the value of the measurement has a timestamp that is greater than the timestamp of the measurement value of the last event for the given IoT device, then replace the measurement value in the stateful data structure and recalculate the average. This stateful operation is a generalization of \"reduce\" and allows one to perform a complex stateful business logic operation and produce a single but different output for each input. This processing produces a stream of key-value pairs, where the key contains information about the beginning and end of the time window. At the same time, the value, known as \"AverageMeasurementValueAggregate\", includes, among other things, the calculated average and all instances of \"SensorTelemetryMeasurementEvent\" that contributed to the calculation. Essentially, each record of this flow reflects a window with the corresponding data. Hopping time windows are aligned to the epoch (Unix timestamp), with the lower interval bound being inclusive and the upper bound being exclusive. **5. suppress**: The abstraction prescribes the deletion of every intermediate change once the grace period is over. That is, only the final result for each time window will be emitted. Comment (limitation): Even after suppress operator is applied, the next event is required to advance the stream time to occur the final result. **6. mapValues**: After a final result is emitted for a time window, it will be transformed into a Meaningful Event. More specifically, the new event includes the start and the end timestamps of the time window, the microservice identifier, the name and the unit of the physical quantity, the calculated average value, the instances of the \"SensorTelemetryMeasurementEvent\" that contributed to the calculation and last, but not least, information about the parameters of the instantiation of the microservice. Technically, the \"AverageMeasurementValueAggregate\" will be mapped to \"WindowedAverageMeasurementValueEvent\", and then the \"WindowedAverageMeasurementValueEvent\" will be enriched with the available information. **7. selectKey**: See the source code and its comments for more information. **8. to**: Each of the generated events, now Meaningful Events, are pushed to \"win-dowed_average_[NAME]_measurement_value_event_[MICRO_SERVICE_ID]\" topic, where \"[NAME]\" and \"[MICRO_SERVICE_ID]\" are the dynamic portions of the topic name and are replaced by the name of the physical quantity and the identifier of the microservice, respectively. The name of the physical quantity is included in the parameter set. At the same time, the microservice ID is automatically generated from the combination of the literal values of the parameters, i.e., a composite key. The result of the final stage is a new event stream with records of type \"WindowedAverageMeasurementValueEvent\". To deal with the relative complexity induced by the dynamic creation of topic names for each instantiation of the proposed abstraction, we propose the gradual merging of event streams of the corresponding topics as follows:  
1. each physical quantity, creation of a topic \"windowed_average_[NAME]_measurement ent_value_event\" where \"[NAME]\" is the dynamic portion replaced by the name of the quantity. All the event streams generated by the microservices related to the specific quantity are merged in this topic. 2. Creation of a topic \"windowed_average_measurement_value_event\" in which all the event streams generated from the previous step are merged. These three abstraction implementations can be instantiated simultaneously and in parallel. Although there is a logical sequence between them, this does not limit the fact that they can operate independently. Thus, the flexibility provided by the proposed implementation of the application is demonstrated. Communication through events is based on conventions, while their processing is based on contracts.