Struct MessageBody

Source

pub struct MessageBody {
    pub typ: String,
    pub msg_id: u64,
    pub in_reply_to: u64,
    pub extra: Map<String, Value>,
}

Expand description

MessageBody represents the reserved keys for a message body.

Fields§

§typ: String

Message type.

§msg_id: u64

Optional. Message identifier that is unique to the source node.

§in_reply_to: u64

Optional. For request/response, the msg_id of the request.

§extra: Map<String, Value>

All the fields not mentioned here

Implementations§

Source §

impl MessageBody

Source

pub fn new() -> Self

Source

pub fn with_type(self, typ: impl Into<String>) -> Self

Examples found in repository ?

examples/echo.rs (line 26)

24    async fn process(&self, runtime: Runtime, req: Message) -> Result<()> {
25        if req.get_type() == "echo" {
26            let echo = req.body.clone().with_type("echo_ok");
27            return runtime.reply(req, echo).await;
28        }
29
30        done(runtime, req)
31    }

Source

pub fn raw(&self) -> Value

use maelstrom::protocol::Message;
use serde_json::Error;

#[derive(serde::Deserialize)]
struct BroadcastRequest {}

fn parse(m: Message) -> Result<BroadcastRequest, Error> {
    serde_json::from_value::<BroadcastRequest>(m.body.raw())
}

Source

pub fn as_obj<'de, T>(&self) -> Result<T>
where T: Deserialize<'de>,

use maelstrom::Result;
use maelstrom::protocol::Message;
use serde_json::Error;

#[derive(serde::Deserialize)]
struct BroadcastRequest {}

fn parse(m: Message) -> Result<BroadcastRequest> {
    m.body.as_obj::<BroadcastRequest>()
}

Examples found in repository ?

examples/lin_kv.rs (line 32)

30    async fn process(&self, runtime: Runtime, req: Message) -> Result<()> {
31        let (ctx, _handler) = Context::new();
32        let msg: Result<Request> = req.body.as_obj();
33        match msg {
34            Ok(Request::Read { key }) => {
35                let value = self.s.get(ctx, key.to_string()).await?;
36                return runtime.reply(req, Request::ReadOk { value }).await;
37            }
38            Ok(Request::Write { key, value }) => {
39                self.s.put(ctx, key.to_string(), value).await?;
40                return runtime.reply(req, Request::WriteOk {}).await;
41            }
42            Ok(Request::Cas { key, from, to, put }) => {
43                self.s.cas(ctx, key.to_string(), from, to, put).await?;
44                return runtime.reply(req, Request::CasOk {}).await;
45            }
46            _ => done(runtime, req),
47        }
48    }

More examples

Hide additional examples

examples/broadcast.rs (line 36)

35    async fn process(&self, runtime: Runtime, req: Message) -> Result<()> {
36        let msg: Result<Request> = req.body.as_obj();
37        match msg {
38            Ok(Request::Read {}) => {
39                let data = self.snapshot();
40                let msg = Request::ReadOk { messages: data };
41                return runtime.reply(req, msg).await;
42            }
43            Ok(Request::Broadcast { message: element }) => {
44                if self.try_add(element) {
45                    info!("messages now {}", element);
46                    for node in runtime.neighbours() {
47                        runtime.call_async(node, Request::Broadcast { message: element });
48                    }
49                }
50
51                return runtime.reply_ok(req).await;
52            }
53            Ok(Request::Topology { topology }) => {
54                let neighbours = topology.get(runtime.node_id()).unwrap();
55                self.inner.lock().unwrap().t = neighbours.clone();
56                info!("My neighbors are {:?}", neighbours);
57                return runtime.reply_ok(req).await;
58            }
59            _ => done(runtime, req),
60        }
61    }

examples/g_set.rs (line 32)

31    async fn process(&self, runtime: Runtime, req: Message) -> Result<()> {
32        let msg: Result<Request> = req.body.as_obj();
33        match msg {
34            Ok(Request::Read {}) => {
35                let data = to_seq(&self.s.lock().unwrap());
36                return runtime.reply(req, Request::ReadOk { value: data }).await;
37            }
38            Ok(Request::Add { element }) => {
39                self.s.lock().unwrap().insert(element);
40                return runtime.reply(req, Request::AddOk {}).await;
41            }
42            Ok(Request::ReplicateOne { element }) => {
43                self.s.lock().unwrap().insert(element);
44                return Ok(());
45            }
46            Ok(Request::ReplicateFull { value }) => {
47                let mut s = self.s.lock().unwrap();
48                for v in value {
49                    s.insert(v);
50                }
51                return Ok(());
52            }
53            Ok(Request::Init {}) => {
54                // spawn into tokio (instead of runtime) to not to wait
55                // until it is completed, as it will never be.
56                let (r0, h0) = (runtime.clone(), self.clone());
57                tokio::spawn(async move {
58                    loop {
59                        tokio::time::sleep(Duration::from_secs(5)).await;
60                        debug!("emit replication signal");
61                        let s = h0.s.lock().unwrap();
62                        for n in r0.neighbours() {
63                            let msg = Request::ReplicateFull { value: to_seq(&s) };
64                            drop(r0.send_async(n, msg));
65                        }
66                    }
67                });
68                return Ok(());
69            }
70            _ => done(runtime, req),
71        }
72    }