Add a `Receiver.new_receiver()` method

[llucax]

What's needed?

A way to, given a receiver, create a new receiver to receive from the same channel or source.

Proposed solution

Add a Receiver.new_receiver() method that will basically create a new receiver in the same channel from the original receiver, with the same configuration of the receiver. It is basically a forward to channel.new_receiver(), when a channel even exists. The default implementation will be to forward all messages received by the original receiver to the new receiver too (implemented using Pipe probably). Map and Filter would not use a pipe, but just call new_receiver on their source receiver.

Use cases

When working with actors, it is sometimes useful to split big tasks into smaller sub-actors. In this case, the parent actor needs to feed these sub-actors with receivers that comes from the receivers passed to the parent actor (for example when listening to dispatches, the subactors might need to listen to dispatches too).

In this case, the parent actor (which normally don't have access to the original channel) needs to create new channels to forward the messages to the sub-actors.

[shsms]

It is basically a forward to channel.new_receiver(), when a channel even exists.

The word fork suggests an Anycast like behaviour. If we want to forward to channel.new_receiver(), I'd also call the function receiver.new_receiver().

Other receivers can implement this differently. For example, a Timer can just create a new timer with the same parameters.

In these cases, it is more expensive to have an extra timer or a file watcher than an extra task and a channel (aka Pipe). In those cases, I'd rather raise with a message saying there's no underlying channel from which to get a new receiver and maybe a hint to use a Pipe.

In the case of a Merge, I'd say it is nicer to use a Pipe rather than cloning all the inner receivers and duplicate the merge operation, but we probably don't have to raise on new_receiver, unless any of the inner receivers raise.

[llucax]

For me receiver.new_receiver() is also confusing. Maybe clone()?

In these cases, it is more expensive to have an extra timer or a file watcher than an extra task and a channel (aka Pipe). In those cases, I'd rather raise with a message saying there's no underlying channel from which to get a new receiver and maybe a hint to use a Pipe.

Yeah, and I think creating a new timer is probably not semantically correct, as it is not really tied to the original receiver's source (if the original receiver is closed, then the cloned receiver should be stopped too and it isn't), so I think the implementation of timer could actually be done using Pipe. I agree for Merge too.

[shsms]

We discussed this and decided to call the method receiver.new_receiver().

We also decided to automatically create a Pipe internally when creating a Timer, FileWatcher or Merge so that receiver.new_receiver calls on them just go to the underlying Channel's receiver.

Map and Filter would not use a pipe, but just call new_receiver on their source receiver.

[llucax]

Updated the title and description.

[shsms]

It sounds complicated for the Timer and FileWatcher.

For example, what would Timer(...) return? If it is a Timer instance that doesn't behave like a receiver, and users have to call new_receiver() at least once, then that sounds possible, but it would be a breaking change.

If we want to retain the current semantics, we will run into a problem—we have to find a way to close the Pipe when all the subscribing receivers have been closed.

[llucax]

Not sure what it is "it" when you say "It sounds complicated for ...".

Anyway, just for the records, @ela-kotulska-frequenz ran into issues while trying to implement the Pipe approach: to use a Pipe we need to create a channel too, and to closing a channel (when we need to close the new piped receiver) is an async operation, so we either need to schedule it in the loop in the receiver close or we need to make receivers have an async close too. Neither sound very sexy.

This made me think a bit more about my original proposal. Here are some comments I made on a private discussion:

I would create a new full receiver. I think the simplest definition of new_receiver() is it is just as creating a new receiver (surprise) with the same configuration of the receiver it is called on. I think this semantic is very clear and can be matched for any receiver

So:

t1 = Timer(a, b, c)
t2 = r.new_receiver()

Is analogous to

t1 = Timer(a, b, c)
t2 = Timer(a, b, c)

And:

c = Broadcast(...)
r1 = c.new_receiver(size=1)
r2 = r1.new_receiver()

Is analogous to

c = Broadcast(...)
r1 = c.new_receiver(size=1)
r2 = c.new_receiver(size=1)

For me this makes sense, it is a matter of ownership. So I get a receiver, and I want another receiver that receives the same stuff, but I want to have control over its life, so if someone closes the original receiver, mine is still alive. I think this is why I called it fork() originally. Is also sort of like duplicating a file descriptor.

@ela-kotulska-frequenz commented about this:

I think it was not clear when new timer should has first tick

• at the same time as original? (first tick should be delayed).
• when it is created (then they will be unsynchronized)

For me it should be as if I called the same constructor used to create the original receiver. I don't think they need to be synchronized if, for example, it is aligned to "now". I think it is better to have a very clear definition that makes (a reasonable amount of) sense to most receivers, than having one separate definition for each receiver. Because again, IMHO the thing we need to care most if we put this into Receiver, is supporting generic code, and for that we need the same definition for all receivers.

All of that said, maybe this proposal still have some holes. I feel I would need some time to think about the implications for every type of receiver to make a sound proposal, but I didn't have the time to do that yet.