@AdriaanPrinsloo
There's two main approaches. The first I would recommend would be to handle your ZMQ IO separately from Riker. You could use Tokio for this or any other IO optimized implementation. When you receive your Request 0mq message you can offload the processing of that to Riker by sending to the respective actor. Your ZMQ queue can continue to handle other Requests. The actor handling the message can choose to send a Response. Typically you could either send a ZMQ reference along with your original message, or you could include such a reference when you create/start the actor. You end up with two "reams", ZMQ and Riker, and no blocking would be necessary.
The other approach would be to handle the ZMQ orchestration and IO inside of Riker. To do this in a non-blocking manner you'd typically use an actor that contains the ZMQ socket and instead of blocking, schedule a message (Check, or Tick, for example) 50ms at the end of recv to be send to myself. Every Tick you check if the ZMQ socket has data and if so get the message. A simple implementation is easy, but if you want to handle a lot of IO then there are patterns you can use to optimize this. The only time you want that Tick message scheduled is when there is no more ZQM messages.
Ideally, you don't ever want to block. Also note that Ask is a future and will require you to block (await) to get the result to the current thread - therefore it's better suited to futures composition where you have a chain of futures working to completion.
(tokio::)select! inside a ctx.run(async move {}) that would wait for some futures including a tokio::delay_for ... of course it wasn't going to just work 😆 I realized Riker comes with its own non tokio executor and I would have to find a way to do my delay future. But made me wonder about how hard it would be for Riker to not spawn futures on its own and like async libraries should do nowadays give the use the choice of what executor to use. Going beyond that probably would be more interesting to have the system better integrated with the whole async/await story, right now it doesn't feel so smooth, a bit like a forced add-on 🤔 it would be cool if all components are future based and the actor system be one big future that is awaited on the main function.
This is a very good point you bring up. Riker isn't using it's own executor, it's using futures::executor::ThreadPool from Futures crate. However, it is using its own instance of the Threadpool executor and futures in a futures chain can't run across different instances. There's also the issue of different versions and implementations of executors. I imagine a way to make Riker (and other application components) share the same executor would be to pass a reference to the executor. In Riker's case, this could be done using:
let sys = SystemBuilder::new()
.name("my-app")
.executor(exec.clone()) // <-- pass the executor reference
.create()
.unwrap();This of course would depend on other libraries allowing for something similar, or getting a reference from a default executor. There wasn't a standard practice for this last time I looked, but that might have changed recently. In any case, I think this is a priority. Appreciate you raising this point.
Hi team, I'm attempting to understand how i can use a BasicActorRef to send one of the Receive implemented messages to. I've modified the messaging.rs example application to illustrate my dilemma: ```
fn main() {
let sys = ActorSystem::new().unwrap();
let actor = sys.actor_of_args::<Counter, _>("counter", 0).unwrap();
actor.tell(Add, None);
let basic_actor_ref : BasicActorRef = actor.clone().into();
match basic_actor_ref.try_tell(Add, None) {
Ok(_) => println!("Success"),
Err(_) => println!("Failure")
}
sys.print_tree();
// force main to wait before exiting program
std::thread::sleep(Duration::from_millis(500));}
```
The actor proc macro creates an internal enum with values for each given type.
#[actor(Add, Sub, Print)]
struct Counter {
count: u32,
}creates the following type:
enum CounterMsg {
Add(Add),
Sub(Sub),
Print(Print),
}This is the type that is supported by the actor. If you use a typed actor reference you can use Add directly, because the macro implements Into<Counter> for the type, and because the type is known, into() works how it should.
If you use a BasicActorRef, the conversion doesn't happen. In this case riker fails to enqueue the message.
match basic_actor_ref.try_tell(CounterMsg::Add(Add), None) {
Ok(_) => println!("Success"),
Err(_) => println!("Failure"),
}
Add would not only add 1, but, whatever we tell it to, maybe even 3!
recv function https://github.com/olanod/riker-es/blob/cqrs/src/entity.rs#L61 I thought ctx.run would run the future already but seems it doesn't and so far the only way I see is doing block_on but I run into the error cannot executeLocalPoolexecutor from within another executor I guess block_on is not meant to be run inside recv? is it because of my #[tokio::test]?
The
ActorSystemis the heart of a Riker application, starting several threads when it is created. Create only one instance ofActorSystemper application.
so, if i create the ActorSystem in main(), how do i access it in a different part of the application?
ctx.system
ActorSystem in main() there's no way to get an ActorRef in some disparate part of my application, other than passing an actor or a system, or a reference to those, into that part of the application
.select on Context and System, which is helpful for interacting with actors through system messages and in some cases user-level messages.
Context, but you are free to pass ActorRef, ActorSystem, etc if you have parts of an application that are not an actor themselves
Sender isn't part of the ActorSystem i need to pass around a ref, just like with the database connection, and, in my naïvité, i thought i could avoid that, but knowing the actor's name
typed function but needs an ExtendedCell as parameter, no idea where to get that, looks like an internal type
