Proposal: add timeout and key options to Stream.partition()

[roveo]

I've been using a custom stream for some time. It works the same way as partition, but with two differences:

1. You can specify timeout option - the number of seconds after which the stream will force-flush the buffer, even if number of elements in it are below the limit.

   This can be useful when you have some data delivery time requirement, but data flow rate is not constant.

2. You can specify key option - a callable that will return a partitioning key from a message, effectively splitting the buffer into several independent buffers.

   This is used when you need your batches to be grouped in some manner. I use this to make sure messages in each batch belong to the same source before passing them on to the sink that writes them to an appropriate database table. It's similar to the PartitionRecord processor in NiFi.

I can make a PR, but the questions are:

• should I? does it make sense to add this to core?
• if yes, should this go into partition or some new operator? The standard behavior won't change so it's not breaking, but implementation details will. Someone might rely on partition._buffer being a list, not a defaultdict.

[martindurant]

Yes, I think this type of stream node sounds useful, and would like to see it included.
I am ambivalent about whether to put it under the existing name with same default behaviour, or make a new class.

• I don't think users ought to be depending on the internal state attribute
• the existing name "partition" perhaps doesn't fully describe the set of options that would be available, but does describe the current simple approach
• the namespace of Stream.attr could get cluttered, if there are multiple implementations that do similar but specialised things.