Add runIf to runEach

[DriesSchaumont]

Describe your changes

Add runif to runEach

Related issue(s)

Closes #xxxx

Type of Change

• [ ] Breaking change (fix or feature that would cause existing functionality to not work as expected)
• [x] New functionality (non-breaking change which adds functionality)
• [ ] Major change (non-breaking change which modifies existing functionality)
• [ ] Minor change (non-breaking change which does not modify existing functionality)
• [ ] Bug fix (non-breaking change which fixes an issue)
• [ ] This change requires a documentation update

Checklist

Requirements:

• [x] I have read the CONTRIBUTING doc.
• [x] I have performed a self-review of my code by checking the "Changed Files" tab.
• [x] My code follows the code style of this project.

Tests:

• [ ] I have added tests that prove my fix is effective or that my feature works.
• [ ] New and existing unit tests pass locally with my changes.

Documentation:

• [x] Proposed changes are described in the CHANGELOG.md.
• [ ] I have updated the documentation accordingly.

Test Environment

[codecov[bot]]

Codecov Report

All modified and coverable lines are covered by tests :white_check_mark:

Project coverage is 92.46%. Comparing base (f109027) to head (504e694). Report is 50 commits behind head on develop.

:exclamation: Current head 504e694 differs from pull request most recent head e5ee18f

Please upload reports for the commit e5ee18f to get more accurate results.

Additional details and impacted files

```diff @@ Coverage Diff @@ ## develop #660 +/- ## =========================================== - Coverage 92.62% 92.46% -0.16% =========================================== Files 128 127 -1 Lines 3917 3889 -28 Branches 631 623 -8 =========================================== - Hits 3628 3596 -32 - Misses 289 293 +4 ```

:umbrella: View full report in Codecov by Sentry.
:loudspeaker: Have feedback on the report? Share it here.

[rcannood]

@DriesSchaumont Bump ;)

[DriesSchaumont]

@rcannood Note that the order in which filter and runIf are executed is the same as for the regular .run: first filter and runIf afterward. Also, the fact that the runIf is also evaluated within the .collect means that the events still get multiplied times the number of input components for .runEach (an alternative implementation is to evaluate the runIf before the collect, not duplicating the events). The way it is implemented now makes it harder to just 'not touch' and event and let is pass through. Thoughts?

[DriesSchaumont]

@rcannood Note that the order in which filter and runIf are executed is the same as for the regular .run: first filter and runIf afterward. Also, the fact that the runIf is also evaluated within the .collect means that the events still get multiplied times the number of input components for .runEach (an alternative implementation is to evaluate the runIf before the collect, not duplicating the events). The way it is implemented now makes it harder to just 'not touch' and event and let is pass through. Thoughts?

Interesting point! ... I guess that it might not be a bad idea to swap runIf and filter around? **Breaking change alert ⚠️

At any rate, I always found it a bit dubious what the expected order of runIf and filter is. Perhaps it would have been better if filter would have had type (id: string, state: map) -> (true | false | "remove" | "skip" | "keep") or something like that.

WRT the breaking change: because runIf is not implemented yet for runEach, I guess its not strictly a breaking change 🤔 But the order being different for runEach and run is not ideal...

I like the idea to expand the filter functionality, but oh boy, talking about breaking changes 😅

P.S.: the duplicated items can be avoided by doing smth like:

def components = [a, b, c]

out = input_ch
  | runEach(
   ....
   filter: {id, state, component ->  component.name == components[0].name}
   ...
  ) 

Its a bit of a hack, but it works