Take Actix as an example to explore how Rust extension features are decoupled in engineering
The opening
Actix is the Rust language implementation of the Actor model
The Actor model is a mathematical model for concurrent computation
The Actor model in 10 minutes
If you look at the Actix source code, you’ll see this code in the Actix :: Handler
// ...
impl<A, M, I, E> MessageResponse<A, M> for Result<I, E>
where
A: Actor,
M: Message<Result = Self>,
I: 'static,
E: 'static,
{
fn handle(self, _ : &mut A::Context, tx: Option<OneshotSender<Self>>) {
tx.send(self)}}// ...
Copy the codeThe implementation is simple. Let’s see what happens.
The type of tx is Option
>. OneshotSender is an alias for tokio::sync::Sender.
The.send(self) method is defined under the actix:: Handler module. The source code is as follows:
// Helper trait for send one shot message from Option<Sender> type.
// None and error are ignored.
trait OneshotSend<M> {
fn send(self, msg: M);
}
impl<M> OneshotSend<M> for Option<OneshotSender<M>> {
fn send(self, msg: M) {
if let Some(tx) = self {
let_ = tx.send(msg); }}}Copy the codeWhere the generic M refers to tokio::sync::Sender
.
Why do we need to define more of this method?
A look at the source code shows that MessageResponse has been implemented a total of 28 times. In other words, the above code uniformly handles the judgment of Option in all Handle functions, reducing the code coupling.
Why not do it some other way?
Let’s take a look at two attempts to see why this approach is better.
Try 1: The helper function
Assuming a helper function is used, the function looks something like this:
fn send<M>(tx: Option<OneshotSender<M>>, msg: M) {
if let Some(tx) = tx {
let_ = tx.send(msg); }}Copy the codeSeems like less code and simpler?
When you use it, you just send(tx, MSG).
But this has the following disadvantages:
- The code is scattered, so if you don’t look at the function definition, you won’t even know you can use it, and it’s easy to have a lot of duplicate code in team collaboration. And therefore, for future refactoring.
- The lack of semantic function expression is also detrimental to team collaboration.
Attempt 2: macro_rule declaration macros
If implemented using declarative macros, the effect would be brutal:
macro_rules! send {
($tx:expr, $msg:expr) => {
if let Some(tx) = $tx {
let_ = $tx.send($msg); }}; }Copy the codeWhile compile time doesn’t stop you here, no one can guess what type $tx and $MSG are unless you write them yourself.
Also, using macro_rule to do this is like an anti-aircraft gun shooting at a mosquito 🙂
Why is the Helper trait better?
Traits, a more powerful implementation of Rust’s specification, can reuse multiple types using generics and can participate in Trait bound as a type constraint.
In order to prevent dependency hell, Rust introduced the Orphan Rule in 2015, which has some limitations. In short, traits and struct/ enums must have one of their own Crate.
For the Orphan rule, see: Little Orphan Impls
Practice – appetizer
As we all know, built-in types are also external. To extend methods for primitive types, you can write traits locally, for example:
trait ColorExtend {
fn is_color(&self) - >bool;
}
Copy the codeWe wrote a color extension method to determine whether the type can be expressed as a color.
Let’s implement this for &str:
// make matching match char
#! [feature(exclusive_range_pattern)]
impl ColorExtend for &str {
fn is_color(&self) - >bool {
if self.len() == 4 || self.len() == 7 {
for elem in self.char_indices().next() {
match elem {
(0.The '#') = > (), (0, _) = >return false(_,'0'.'9') = > (), (_,'a'.'z') = > (), (_,'A'.'Z') => (),
_ => return false,}}true
} else {
false}}}Copy the codeAfter that we can use it like this: “#0000FF”.is_color()
Similarly, this method can be implemented for other types, such as String, Vec
, etc., or even for Option<& STR >. All generic types belong to a single type. Due to Rust’s zero-cost abstraction, the compiler expands generics to generate separate types. With that in mind, let’s start extending some of the types in Actix.
Write the Actor
Define an actor for counting as an example and implement some basic characteristics for it:
use actix::{Actor, Context};
pub struct MyActor {
pub counter: u16,}impl Default for MyActor {
fn default() - >Self {
Self { counter: Default::default() }
}
}
impl Actor for MyActor {
type Context = Context<Self>;
}
Copy the codeGet the countGetCounter
use actix::{Handler, Message};
/// counter add message
#[derive(Debug, Message)]
#[rtype("u16")]
pub struct GetCounter;
impl Handler<GetCounter> for MyActor {
type Result = u16;
fn handle(&mut self, _: GetCounter, _: &mut Self::Context) -> Self::Result {
self.counter
}
}
Copy the codecountCounterAdd
/// counter add message
#[derive(Debug, Message)]
#[rtype("u16")]
pub struct CounterAdd(pub u16);
impl Handler<CounterAdd> for MyActor {
type Result = u16;
fn handle(&mut self, msg: CounterAdd, _: &mut Self::Context) -> Self::Result {
println!("add {}", msg.0);
self.counter += msg.0;
self.counter
}
}
Copy the codeThe count changes in n secondsGetDelta
Here we return ResponseActFuture because we need asynchrony.
use std::time::Duration;
/// get counter's value change during the [`Duration`]
#[derive(Debug, Message)]
#[rtype("u16")]
pub struct GetDelta(pub Duration);
impl Handler<GetDelta> for MyActor {
type Result = ResponseActFuture<Self.u16>;
fn handle(&mut self, msg: GetDelta, _: &mut Self::Context) -> Self::Result {
let init_value = self.counter; / / initial value
Box::pin(
async move {
actix::clock::sleep(msg.0).await;
}
.into_actor(self)
.map(move |_, actor, _| {
// Wait for the future to end and get the latest counter, subtracting the original value
actor.counter - init_value
})
)
}
}
Copy the codeActix :: Clock ::sleep does not block because Actix uses a Tokio based runtime
Here is actix into_actor method: : fut… the future: : WrapFuture defined in this paper, the proposed helper trait, this idea is very popular in some libraries.
WrapFuture definition:
pub trait WrapFuture<A>
where
A: Actor,
{
/// The future that this type can be converted into.
type Future: ActorFuture<A>;
/// Convert normal future to a ActorFuture
fn into_actor(self, a: &A) -> Self::Future;
}
impl<F: Future, A: Actor> WrapFuture<A> for F {
type Future = FutureWrap<F, A>;
fn into_actor(self, _: &A) -> Self::Future {
wrap_future(self)}}Copy the codeHere we add an into_actor method for the Future that returns a FutureWrap, which happens to be defined using the pin_project I mentioned last time.
Why does Rust need Pin, Unpin?
To simulate the demand
Let’s say the product manager has a requirement to count the change in the value of counter in n seconds, and then add an additional equivalent value after n seconds.
/// an odd mission required by the lovely PM
#[derive(Debug, Message)]
#[rtype("()")]
pub struct DoubleAfterDelta {
pub secs: u64
}
impl Handler<DoubleAfterDelta> for MyActor {
type Result = ResponseActFuture<Self, () >;fn handle(&mut self, msg: DoubleAfterDelta, ctx: &mut Self::Context) -> Self::Result {
Box::pin({
let addr = ctx.address();
addr.send(GetDelta(
Duration::from_secs(msg.secs)
))
.into_actor(self)
.map(move |ret, actor, ctx| {
// Manually added functions
ret.handle_mailbox(|delta| {
// This is also a manually added functionctx.add_later(delta, msg.secs); }); }}})})Copy the codeFor the parameter function of map, the three parameters are of type:
-
Result<T, MailboxError> -
&mut MyActor -
&mut Context<MyActor>
Assuming we write it in plain form for the moment, it would look something like this:
match ret {
Ok(data) => ctx.notify_later(CounterAdd(data), Duration::from_secs(msg.secs)),
Err(e) => eprintln! ("common handle MailboxError: {}", e),
}
Copy the codeEmmm… It seems all right.
Result
and &mut Context
. Result
and &mut Context
Let’s pull this part of the code out.
encapsulationResult<T, MailboxError>
First, define a trait:
pub trait ActixMailboxSimplifyExtend<T> {
fn handle_mailbox<F>(self, handle_fn: F)
where
F: FnOnce(T) -> ();
}
Copy the codeThis method receives a closure called handler, which handles only normal returns and MailboxError in handle_mailbox, as follows:
impl<T> ActixMailboxSimplifyExtend<T> for Result<T, MailboxError> {
fn handle_mailbox<F>(self, handle_fn: F)
where
F: FnOnce(T) -> () {
match self {
Ok(data) => handle_fn(data),
Err(e) => eprintln! ("common handle MailboxError: {}", e),
}
}
}
Copy the codeencapsulationContext<MyActor>
pub trait ActixContextExtend {
fn add_later(&mut self, add_num: u16, secs: u64) - > (); }impl ActixContextExtend for Context<MyActor> {
fn add_later(&mut self, add_num: u16, secs: u64) - > () {println!("counter will add {}, after {} second(s)", add_num, secs);
self.notify_later(CounterAdd(add_num), Duration::from_secs(secs)); }}Copy the codeThe add_later function encapsulates notify_later and automatically sends CounterAdd messages after secs.
Having encapsulated these two approaches, you Don’t Repeat Yourself when you encounter similar requirements.
defects
Of course, this approach has its drawbacks and requires manual introduction of declarations. The compiler does not look for all of its implementations because of dependency confusion. If two methods are repeated, the compiler will report the following error:
error[E0034]: multiple applicable items in scope
Copy the codeSo try to avoid naming functions that overlap with existing ones, or refactoring can get messy.
The main function
So much for the setup, let’s look at the actual results.
fn main() {
let run_ret = actix::run(async move {
let actor = MyActor::default();
let addr = actor.start();
println!("=[case 1]=========================");
let current_value = addr.send(GetCounter).await.unwrap();
println!("init value is: {}", current_value);
let fut = addr.send(DoubleAfterDelta {
secs: 1});// add during DoubleAfterDelta's Handler waiting
sleep(Duration::from_millis(200)).await; // actix::clock::sleep
addr.do_send(CounterAdd(3));
sleep(Duration::from_millis(200)).await;
addr.do_send(CounterAdd(5));
let _ = fut.await; // wait a seconds.
let current_value = addr.send(GetCounter).await.unwrap();
println!("value is: {}", current_value);
sleep(Duration::from_secs(2)).await;
let current_value = addr.send(GetCounter).await.unwrap();
println!("value is: {}", current_value);
println!("=[case 2]=========================");
addr.do_send(ShutDown); // Close the actor message
let ret = addr.send(GetCounter).await;
// use the added method in ActixMailboxSimplifyExtend
ret.handle_mailbox(|_| {
unreachable!("unpossible to reach here due to MailboxError must be encountered.");
});
});
println!("actix-run: {:? }", run_ret);
}
Copy the codeThis tests the availability of the add_later function and whether handle_mailbox handles errors correctly.
Running results:
=[case 1]=========================
init value is: 0
add 3
add 5
counter will add 8, after 1 second(s)
value is: 8
add 8
value is: 16
=[case 2]=========================
common handle MailboxError: Mailbox has closed
actix-run: Ok(())
Copy the codeconclusion
Based on years of language experience, the founders of Rust summed up the experience of previous developers and advocated composition over inheritance. Use combinations to extend your code and make your code Rusty!
This article has been pushed to Github, if you are inspired to welcome star.
Github.com/oloshe/rust…
Reprint please state the source