feat: Add dropRepeated stream operator.

[He-Pin]

Motivation:
Add dropRepeated stream operator.

refs: #1662

[He-Pin]

@pjfanning Would you like to give another round of review on this? I want to add aggregateUntilChanged once this is merged.

[pjfanning]

Is there a reason not to just call this distinct? I don't get what the UntilChanged bit adds to this - ie it seems unnecessary and potentially confusing.

[raboof]

Is there a reason not to just call this distinct? I don't get what the UntilChanged bit adds to this - ie it seems unnecessary and potentially confusing.

I think we should try and avoid confusion with https://www.scala-lang.org/api/3.7.0/scala/collection/SeqOps.html#distinct-0

I understand the name follows https://rxjs.dev/api/operators/distinctUntilChanged . I'm not sure it makes sense to me: the results aren't 'distinct' 'until' they are the same? I don't want to bikeshed too much, but something like `dropRepeated' might make more sense? We can mention in the docs that this is called distinctUntilChanged in rx so people can find it.

The docs for 'statefulMap' have this as an example (https://pekko.apache.org/docs/pekko/current/stream/operators/Source-or-Flow/statefulMap.html). There's something to be said for having it as a separate construct, but perhaps the docs should link to each other?

[He-Pin]

@pjfanning I'm following the naming in reactor-core and smallrye/smallrye-mutiny#1853, rxjava is using this name too.

The main reason to add more operator is because when reading the code of Spring-Ai, and I found these operators are quite common, I hope someone will choose pekko to build scala's Spring-Ai

And we are building our mcp server at work too:)

[pjfanning]

Is there a reason not to just call this distinct? I don't get what the UntilChanged bit adds to this - ie it seems unnecessary and potentially confusing.

I think we should try and avoid confusion with https://www.scala-lang.org/api/3.7.0/scala/collection/SeqOps.html#distinct-0

I understand the name follows https://rxjs.dev/api/operators/distinctUntilChanged . I'm not sure it makes sense to me: the results aren't 'distinct' 'until' they are the same? I don't want to bikeshed too much, but something like `dropRepeated' might make more sense? We can mention in the docs that this is called distinctUntilChanged in rx so people can find it.

The docs for 'statefulMap' have this as an example (https://pekko.apache.org/docs/pekko/current/stream/operators/Source-or-Flow/statefulMap.html). There's something to be said for having it as a separate construct, but perhaps the docs should link to each other?

I suppose that this function is not then the same as distinct and I think dropRepeated makes a lot more sense than distinctUntilChanged

[He-Pin]

@pjfanning @pjfanning And in Kotlin flow, it's distinctUntilChanged ,but in FS2, it's filterWithPrevious.

and changes and changesWith in ZIO

Which name do you prefer, actually, I don't like the distinctUntilChanged either:)

[raboof]

I could live with changes or dropRepeated. (I don't think we need an equivalent of changesWith, people can use statefulMap for more advanced use cases)

[He-Pin]

@pjfanning @raboof I have update to dropRepeated

[He-Pin]

@pjfanning Thanks

[pjfanning]

lgtm