How to drain the queue of messages without stop the actor?

[jBernavaPrah]

Hi! First of all, thank you for this crate! :)

I'm trying to drain all the messages from the actor queue, without stopping it. What is the correct way to do it?

To provide more context:
I have an actor connected to a web socket. I send this actor a message to perform some long-running tasks, and the results are sent back via a different channel.

I may have sent many messages to this actor that are still in the queue. Due to some external circumstances, I would like to remove all the remaining messages in the queue without stopping the actor to avoid having to reconnect to the service.

Do you have any suggestions in place?

Happy to provide more context if necessary!
Thanks a lot!
J

[slawlor]

I forget which exact release introduced it but there is a drain() method on the ActorCell.

Try updating the crate to a more recent release?

[jBernavaPrah]

Thanks @slawlor for the reply!

I have already tried with drain() and drain_and_wait(), but both stop the actor after the drain is completed, as mentioned in the documentation.

[slawlor]

Ah ok I missed that. There's no support for this built in today and I'm not exactly sure how you'd do it anyways sooner the request will still be processed.

You would probably need to build a supervisor that supervises a single child. The child only does one request at a time, and the supervisor keeps a queue of all their requests to be processed, this is similar to how a factory works. And ignore the other requests. The supervisor does this since it's not blocking on doing an operation, and the child won't see any future items.

Unfortunately, you need to inspect the queue of messages in this case, so you have to manage the queue yourself in a parent actor

[slawlor]

Ah ok I missed that. There's no support for this built in today and I'm not exactly sure how you'd do it anyways sooner the request will still be processed.

You would probably need to build a supervisor that supervises a single child. The child only does one request at a time, and the supervisor keeps a queue of all their requests to be processed, this is similar to how a factory works. And ignore the other requests. The supervisor does this since it's not blocking on doing an operation, and the child won't see any future items.

Unfortunately, you need to inspect the queue of messages in this case, so you have to manage the queue yourself in a parent actor

[jBernavaPrah]

Thanks a lot for your reply!
I hoped there was something already in place that I did not see.

Anyway, could you point me where I could find something related to the factory you mentioned?

[slawlor]

So the factory primitive is a bit complicated, mainly due to its flexibility but I'll do my best to give pointers to relevant bits. Your use-case is unique and specific, so much of the nature of the "factory" won't be necessary for your usage.

So here is where the factory selects the next "job" for a worker to process. The factory has a generic queueing primitive, so there's many types of queues with different functionality, that you can support. However in your case a simple Vec<TMessage> is probably sufficient.

Then when the factory is told to "drain" requests, internally in its state it changes to a "draining" mode here. After that transition in the state, all future requests are rejected and the remaining queued items are processed (if any) until there's no more requests in flight or in queue.

In your case, I think you'd probably just flush the queue'd message in the factory, dropping them without the "child"/worker processing them. Then allowing the next request to come in. What I'd imagine is

1. Create a Parent -> Child actor setup, where the child has the actual processing logic, and the parent tracks if the child is processing a message or not, and holds a queue of "backlogged' messages. The main idea here is that the parent shouldn't get "blocked" doing anything, so if new messages come in, it can put them into its internal queue. The child will need to notify the parent when it's done a message, so the parent can send the next message (or not) to be processed.
2. The parent would have 2 message variants "process" and "flush". If flush is received, whatever is in the parent's internal queue is just dropped.

I think that's all you'd need to do.