Efficient distinct operator

[kevin-dp]

This PR replaces the existing distinct operator which was built on top of the reduce operator with a more efficient implementation. This new distinct operator supports an optional argument that is a function (value: T) => any to determine what to deduplicate on. For example, we can do:

// Get distinct users
usersInput.pipe(
  distinct()
)

// Get distinct users per country
usersInput.pipe(
  distinct(user => user.country)
)

Note that the output has the same type as the input, the function is only used to determine what to deduplicate on but it is not used to transform the values. This function is very useful for ts/db where we want to get distinct results based on the selected columns but we still want the entire selected rows to come out of the stream.

[samwillis]

I'll take a proper look later when home, but looks good. Does this dedupe per-key? So to dedupe the all you key by a common key first?

[samwillis]

Not a problem if it doesn't, we can always add a version that does if we want to later in.

[kevin-dp]

I'll take a proper look later when home, but looks good. Does this dedupe per-key? So to dedupe the all you key by a common key first?

It dedupes by the entire value unless you pass an extractor function in which case it dedupes by the return value of that function. The extractor function pattern should be general enough to handle any case.

So, if you want to dedup by key you would do distinct(([key, _value]) => key). If you want to dedup by value (across keys) you can do distinct(([_key, value]) => value).

[samwillis]

Awesome!