Struct rings_core::message::MessageRelay

source · 
pub struct MessageRelay {
    pub method: RelayMethod,
    pub path: Vec<Did>,
    pub path_end_cursor: usize,
    pub next_hop: Option<Did>,
    pub destination: Did,
}
Expand description

MessageRelay guide message passing on rings network by relay.

All messages should be sent with MessageRelay. By calling relay method in correct place, MessageRelay help to do things:

  • Infer the next_hop of a message.
  • Get the sender and origin sender of a message.
  • Record the whole transport path for inspection.

Examples

Fields§

§method: RelayMethod

The method of message. SEND or REPORT.

§path: Vec<Did>

A push only stack. Record routes when handling sending messages.

§path_end_cursor: usize

Move this cursor to flag the top of the stack when reporting. Notice that this cursor is not the index of current. It’s path.len() - <index of current> - 1, which means count down to head of vector. It will always be 0 while handling sending messages in this way.

§next_hop: Option<Did>

The next node to handle the message. When and only when located at the end of the message propagation, the next_hop is none. A message handler will pick transport by this field.

§destination: Did

The destination of the message. It may be customized when sending. It cannot be changed when reporting. It may help the handler to find out next_hop in some situations.

Implementations§

Create a new MessageRelay. Will set path_end_cursor to 0 if pass None as parameter.

Examples found in repository?
src/message/payload.rs (lines 115-121)
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123

    pub fn new_send(
        data: T,
        session_manager: &SessionManager,
        next_hop: Did,
        destination: Did,
    ) -> Result<Self> {
        let relay = MessageRelay::new(
            RelayMethod::SEND,
            vec![session_manager.authorizer()?],
            None,
            Some(next_hop),
            destination,
        );
        Self::new(data, session_manager, OriginVerificationGen::Origin, relay)
    }

Check current did, update path and its end cursor, then infer next_hop.

When handling a SEND message, will push current to the self.path stack, and set next_hop parameter to self.next_node.

When handling a REPORT message, will move forward self.path_end_cursor to the position of current in self.path. If next_hop parameter is none, it will also pick the previous node in self.path as self.next_hop. (With this feature, one can always pass None as next_hop parameter when handling a REPORT message.)

Examples found in repository?
src/message/handlers/subring.rs (line 76)
71
72
73
74
75
76
77
78
79
80
81
82
83
84

    async fn handle(&self, ctx: &MessagePayload<Message>, msg: &JoinSubRing) -> Result<()> {
        let mut relay = ctx.relay.clone();
        let origin = relay.origin();
        match self.dht.join_subring(origin, msg.rid).await {
            Ok(PeerRingAction::RemoteAction(next, RemoteAction::FindAndJoinSubRing(_))) => {
                relay.relay(self.dht.did, Some(next))?;
                relay.reset_destination(next)?;
                self.forward_payload(ctx, relay).await
            }
            Ok(PeerRingAction::None) => Ok(()),
            Ok(act) => Err(Error::PeerRingUnexpectedAction(act)),
            Err(e) => Err(e),
        }
    }
More examples
Hide additional examples
src/message/handlers/stabilization.rs (line 29)
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43

    async fn handle(
        &self,
        ctx: &MessagePayload<Message>,
        msg: &NotifyPredecessorSend,
    ) -> Result<()> {
        let mut relay = ctx.relay.clone();
        let predecessor = { *self.dht.lock_predecessor()? };

        relay.relay(self.dht.did, None)?;
        self.dht.notify(msg.did)?;
        if let Some(did) = predecessor {
            if did != relay.origin() {
                return self
                    .send_report_message(
                        Message::NotifyPredecessorReport(NotifyPredecessorReport { did }),
                        ctx.tx_id,
                        relay,
                    )
                    .await;
            }
        }
        Ok(())
    }
src/message/handlers/custom.rs (line 28)
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43

    async fn handle(
        &self,
        ctx: &MessagePayload<Message>,
        _: &MaybeEncrypted<CustomMessage>,
    ) -> Result<()> {
        let mut relay = ctx.relay.clone();

        if self.dht.did != relay.destination {
            if self.swarm.get_transport(relay.destination).is_some() {
                relay.relay(self.dht.did, Some(relay.destination))?;
                return self.forward_payload(ctx, relay).await;
            } else {
                let next_node = match self.dht.find_successor(relay.destination)? {
                    PeerRingAction::Some(node) => Some(node),
                    PeerRingAction::RemoteAction(node, _) => Some(node),
                    _ => None,
                }
                .ok_or(Error::MessageHandlerMissNextNode)?;
                relay.relay(self.dht.did, Some(next_node))?;
                return self.forward_payload(ctx, relay).await;
            }
        }

        Ok(())
    }
src/message/handlers/storage.rs (line 90)
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149

    async fn handle(&self, ctx: &MessagePayload<Message>, msg: &SearchVNode) -> Result<()> {
        let mut relay = ctx.relay.clone();

        match self.dht.lookup(msg.vid).await {
            Ok(action) => match action {
                PeerRingAction::None => Ok(()),
                PeerRingAction::SomeVNode(v) => {
                    relay.relay(self.dht.did, None)?;
                    self.send_report_message(
                        Message::FoundVNode(FoundVNode { data: vec![v] }),
                        ctx.tx_id,
                        relay,
                    )
                    .await
                }
                PeerRingAction::RemoteAction(next, _) => {
                    relay.relay(self.dht.did, Some(next))?;
                    self.forward_payload(ctx, relay).await
                }
                act => Err(Error::PeerRingUnexpectedAction(act)),
            },
            Err(e) => Err(e),
        }
    }
}

#[cfg_attr(feature = "wasm", async_trait(?Send))]
#[cfg_attr(not(feature = "wasm"), async_trait)]
impl HandleMsg<FoundVNode> for MessageHandler {
    async fn handle(&self, ctx: &MessagePayload<Message>, msg: &FoundVNode) -> Result<()> {
        let mut relay = ctx.relay.clone();

        relay.relay(self.dht.did, None)?;
        if relay.next_hop.is_some() {
            self.forward_payload(ctx, relay).await
        } else {
            // When query successor, store in local cache
            for datum in msg.data.iter().cloned() {
                self.dht.local_cache_set(datum);
            }
            Ok(())
        }
    }
}

#[cfg_attr(feature = "wasm", async_trait(?Send))]
#[cfg_attr(not(feature = "wasm"), async_trait)]
impl HandleMsg<StoreVNode> for MessageHandler {
    async fn handle(&self, ctx: &MessagePayload<Message>, msg: &StoreVNode) -> Result<()> {
        let virtual_peer = msg.data.clone();
        for p in virtual_peer {
            match self.dht.store(p).await {
                Ok(action) => match action {
                    PeerRingAction::None => Ok(()),
                    PeerRingAction::RemoteAction(next, _) => {
                        let mut relay = ctx.relay.clone();
                        relay.reset_destination(next)?;
                        relay.relay(self.dht.did, Some(next))?;
                        self.forward_payload(ctx, relay).await
                    }
                    act => Err(Error::PeerRingUnexpectedAction(act)),
                },
                Err(e) => Err(e),
            }?;
        }
        Ok(())
    }
src/message/handlers/connection.rs (line 86)
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265

    async fn handle(&self, ctx: &MessagePayload<Message>, msg: &ConnectNodeSend) -> Result<()> {
        let mut relay = ctx.relay.clone();
        // if id is not dest
        if self.dht.did != relay.destination {
            if self
                .swarm
                .get_and_check_transport(relay.destination)
                .await
                .is_some()
            {
                relay.relay(self.dht.did, Some(relay.destination))?;
                return self.forward_payload(ctx, relay).await;
            } else {
                let next_node = match self.dht.find_successor(relay.destination)? {
                    PeerRingAction::Some(node) => Some(node),
                    PeerRingAction::RemoteAction(node, _) => Some(node),
                    _ => None,
                }
                .ok_or(Error::MessageHandlerMissNextNode)?;
                relay.relay(self.dht.did, Some(next_node))?;
                return self.forward_payload(ctx, relay).await;
            }
        } else {
            // self is dest
            relay.relay(self.dht.did, None)?;
            match self.swarm.get_and_check_transport(relay.sender()).await {
                None => {
                    let trans = self.swarm.new_transport().await?;
                    trans
                        .register_remote_info(msg.handshake_info.to_owned().into())
                        .await?;
                    let handshake_info = trans
                        .get_handshake_info(
                            self.swarm.session_manager(),
                            RTCSdpType::Answer,
                            self.swarm.services(),
                        )
                        .await?
                        .to_string();
                    self.send_report_message(
                        Message::ConnectNodeReport(ConnectNodeReport {
                            transport_uuid: msg.transport_uuid.clone(),
                            handshake_info,
                        }),
                        ctx.tx_id,
                        relay,
                    )
                    .await?;
                    self.swarm.push_pending_transport(&trans)?;
                    Ok(())
                }

                _ => {
                    self.send_report_message(
                        Message::AlreadyConnected(AlreadyConnected),
                        ctx.tx_id,
                        relay,
                    )
                    .await
                }
            }
        }
    }
}

#[cfg_attr(feature = "wasm", async_trait(?Send))]
#[cfg_attr(not(feature = "wasm"), async_trait)]
impl HandleMsg<ConnectNodeReport> for MessageHandler {
    async fn handle(&self, ctx: &MessagePayload<Message>, msg: &ConnectNodeReport) -> Result<()> {
        let mut relay = ctx.relay.clone();

        relay.relay(self.dht.did, None)?;
        if relay.next_hop.is_some() {
            self.forward_payload(ctx, relay).await
        } else {
            let transport = self
                .swarm
                .find_pending_transport(
                    uuid::Uuid::from_str(&msg.transport_uuid)
                        .map_err(|_| Error::InvalidTransportUuid)?,
                )?
                .ok_or(Error::MessageHandlerMissTransportConnectedNode)?;
            transport
                .register_remote_info(msg.handshake_info.clone().into())
                .await?;
            Ok(())
        }
    }
}

#[cfg_attr(feature = "wasm", async_trait(?Send))]
#[cfg_attr(not(feature = "wasm"), async_trait)]
impl HandleMsg<AlreadyConnected> for MessageHandler {
    async fn handle(&self, ctx: &MessagePayload<Message>, _msg: &AlreadyConnected) -> Result<()> {
        let mut relay = ctx.relay.clone();

        relay.relay(self.dht.did, None)?;
        if relay.next_hop.is_some() {
            self.forward_payload(ctx, relay).await
        } else {
            self.swarm
                .get_and_check_transport(relay.sender())
                .await
                .map(|_| ())
                .ok_or(Error::MessageHandlerMissTransportAlreadyConnected)
        }
    }
}

#[cfg_attr(feature = "wasm", async_trait(?Send))]
#[cfg_attr(not(feature = "wasm"), async_trait)]
impl HandleMsg<FindSuccessorSend> for MessageHandler {
    async fn handle(&self, ctx: &MessagePayload<Message>, msg: &FindSuccessorSend) -> Result<()> {
        let mut relay = ctx.relay.clone();

        match self.dht.find_successor(msg.did)? {
            PeerRingAction::Some(did) => {
                if !msg.strict || self.dht.did == msg.did {
                    match &msg.then {
                        FindSuccessorThen::Report(handler) => {
                            relay.relay(self.dht.did, None)?;
                            self.send_report_message(
                                Message::FindSuccessorReport(FindSuccessorReport {
                                    did,
                                    handler: handler.clone(),
                                }),
                                ctx.tx_id,
                                relay,
                            )
                            .await
                        }
                    }
                } else if self.swarm.get_and_check_transport(msg.did).await.is_some() {
                    relay.relay(self.dht.did, Some(relay.destination))?;
                    return self.forward_payload(ctx, relay).await;
                } else {
                    return Err(Error::MessageHandlerMissNextNode);
                }
            }
            PeerRingAction::RemoteAction(next, _) => {
                relay.relay(self.dht.did, Some(next))?;
                relay.reset_destination(next)?;
                self.forward_payload(ctx, relay).await
            }
            act => Err(Error::PeerRingUnexpectedAction(act)),
        }
    }
}

#[cfg_attr(feature = "wasm", async_trait(?Send))]
#[cfg_attr(not(feature = "wasm"), async_trait)]
impl HandleMsg<FindSuccessorReport> for MessageHandler {
    async fn handle(&self, ctx: &MessagePayload<Message>, msg: &FindSuccessorReport) -> Result<()> {
        let mut relay = ctx.relay.clone();

        relay.relay(self.dht.did, None)?;
        if relay.next_hop.is_some() {
            return self.forward_payload(ctx, relay).await;
        }

        match &msg.handler {
            // TODO: how to prevent `fix_finger_index` before got `FixFingerTable`?
            FindSuccessorReportHandler::FixFingerTable => self.dht.lock_finger()?.set_fix(msg.did),
            FindSuccessorReportHandler::Connect => {
                if self.swarm.get_and_check_transport(msg.did).await.is_none()
                    && msg.did != self.swarm.did()
                {
                    self.swarm.connect(msg.did).await?;
                }
            }
            FindSuccessorReportHandler::SyncStorage => {
                self.dht.lock_successor()?.update(msg.did);
                if let Ok(PeerRingAction::RemoteAction(
                    next,
                    PeerRingRemoteAction::SyncVNodeWithSuccessor(data),
                )) = self.dht.sync_with_successor(msg.did).await
                {
                    self.send_direct_message(
                        Message::SyncVNodeWithSuccessor(SyncVNodeWithSuccessor { data }),
                        next,
                    )
                    .await?;
                    return Ok(());
                }
            }
            _ => {}
        }

        Ok(())
    }

Construct a MessageRelay of method REPORT from a MessageRelay of method REPORT. It will return Error if method is not SEND. It will return Error if self.path.len() is less than 2.

Examples found in repository?
src/message/payload.rs (line 131)
125
126
127
128
129
130
131
132
133
134
135

    pub fn new_report(
        data: T,
        tx_id: uuid::Uuid,
        session_manager: &SessionManager,
        relay: &MessageRelay,
    ) -> Result<Self> {
        let relay = relay.report()?;
        let mut pl = Self::new(data, session_manager, OriginVerificationGen::Origin, relay)?;
        pl.tx_id = tx_id;
        Ok(pl)
    }

A SEND message can change its destination. Call with REPORT method will get an error imeediately.

Examples found in repository?
src/message/handlers/subring.rs (line 77)
71
72
73
74
75
76
77
78
79
80
81
82
83
84

    async fn handle(&self, ctx: &MessagePayload<Message>, msg: &JoinSubRing) -> Result<()> {
        let mut relay = ctx.relay.clone();
        let origin = relay.origin();
        match self.dht.join_subring(origin, msg.rid).await {
            Ok(PeerRingAction::RemoteAction(next, RemoteAction::FindAndJoinSubRing(_))) => {
                relay.relay(self.dht.did, Some(next))?;
                relay.reset_destination(next)?;
                self.forward_payload(ctx, relay).await
            }
            Ok(PeerRingAction::None) => Ok(()),
            Ok(act) => Err(Error::PeerRingUnexpectedAction(act)),
            Err(e) => Err(e),
        }
    }
More examples
Hide additional examples
src/message/handlers/storage.rs (line 139)
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149

    async fn handle(&self, ctx: &MessagePayload<Message>, msg: &StoreVNode) -> Result<()> {
        let virtual_peer = msg.data.clone();
        for p in virtual_peer {
            match self.dht.store(p).await {
                Ok(action) => match action {
                    PeerRingAction::None => Ok(()),
                    PeerRingAction::RemoteAction(next, _) => {
                        let mut relay = ctx.relay.clone();
                        relay.reset_destination(next)?;
                        relay.relay(self.dht.did, Some(next))?;
                        self.forward_payload(ctx, relay).await
                    }
                    act => Err(Error::PeerRingUnexpectedAction(act)),
                },
                Err(e) => Err(e),
            }?;
        }
        Ok(())
    }
src/message/handlers/connection.rs (line 217)
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222

    async fn handle(&self, ctx: &MessagePayload<Message>, msg: &FindSuccessorSend) -> Result<()> {
        let mut relay = ctx.relay.clone();

        match self.dht.find_successor(msg.did)? {
            PeerRingAction::Some(did) => {
                if !msg.strict || self.dht.did == msg.did {
                    match &msg.then {
                        FindSuccessorThen::Report(handler) => {
                            relay.relay(self.dht.did, None)?;
                            self.send_report_message(
                                Message::FindSuccessorReport(FindSuccessorReport {
                                    did,
                                    handler: handler.clone(),
                                }),
                                ctx.tx_id,
                                relay,
                            )
                            .await
                        }
                    }
                } else if self.swarm.get_and_check_transport(msg.did).await.is_some() {
                    relay.relay(self.dht.did, Some(relay.destination))?;
                    return self.forward_payload(ctx, relay).await;
                } else {
                    return Err(Error::MessageHandlerMissNextNode);
                }
            }
            PeerRingAction::RemoteAction(next, _) => {
                relay.relay(self.dht.did, Some(next))?;
                relay.reset_destination(next)?;
                self.forward_payload(ctx, relay).await
            }
            act => Err(Error::PeerRingUnexpectedAction(act)),
        }
    }

Check if path and destination is valid. It will be automatically called at relay started.

Examples found in repository?
src/message/protocols/relay.rs (line 81)
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124

    pub fn relay(&mut self, current: Did, next_hop: Option<Did>) -> Result<()> {
        self.validate()?;

        // If self.next_hop is set, it should be current
        if self.next_hop.is_some() && self.next_hop.unwrap() != current {
            return Err(Error::InvalidNextHop);
        }

        match self.method {
            RelayMethod::SEND => {
                self.path.push(current);
                self.next_hop = next_hop;
                Ok(())
            }

            RelayMethod::REPORT => {
                // The final hop
                if self.next_hop == Some(self.destination) {
                    self.path_end_cursor = self.path.len() - 1;
                    self.next_hop = None;
                    return Ok(());
                }

                let pos = self
                    .path
                    .iter()
                    .rev()
                    .skip(self.path_end_cursor)
                    .position(|&x| x == current);

                if let (None, None) = (pos, next_hop) {
                    return Err(Error::CannotInferNextHop);
                }

                if let Some(pos) = pos {
                    self.path_end_cursor += pos;
                }

                // `self.path_prev()` should never return None here, because we have handled final hop
                self.next_hop = next_hop.or_else(|| self.path_prev());

                Ok(())
            }
        }
    }

Get the original sender of current message. Should always be the first element of path.

Examples found in repository?
src/message/protocols/relay.rs (line 190)
188
189
190
191
192
193

    pub fn sender(&self) -> Did {
        match self.method {
            RelayMethod::SEND => self.origin(),
            RelayMethod::REPORT => *self.path.last().unwrap(),
        }
    }
More examples
Hide additional examples
src/message/handlers/subring.rs (line 73)
71
72
73
74
75
76
77
78
79
80
81
82
83
84

    async fn handle(&self, ctx: &MessagePayload<Message>, msg: &JoinSubRing) -> Result<()> {
        let mut relay = ctx.relay.clone();
        let origin = relay.origin();
        match self.dht.join_subring(origin, msg.rid).await {
            Ok(PeerRingAction::RemoteAction(next, RemoteAction::FindAndJoinSubRing(_))) => {
                relay.relay(self.dht.did, Some(next))?;
                relay.reset_destination(next)?;
                self.forward_payload(ctx, relay).await
            }
            Ok(PeerRingAction::None) => Ok(()),
            Ok(act) => Err(Error::PeerRingUnexpectedAction(act)),
            Err(e) => Err(e),
        }
    }
src/message/handlers/stabilization.rs (line 32)
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43

    async fn handle(
        &self,
        ctx: &MessagePayload<Message>,
        msg: &NotifyPredecessorSend,
    ) -> Result<()> {
        let mut relay = ctx.relay.clone();
        let predecessor = { *self.dht.lock_predecessor()? };

        relay.relay(self.dht.did, None)?;
        self.dht.notify(msg.did)?;
        if let Some(did) = predecessor {
            if did != relay.origin() {
                return self
                    .send_report_message(
                        Message::NotifyPredecessorReport(NotifyPredecessorReport { did }),
                        ctx.tx_id,
                        relay,
                    )
                    .await;
            }
        }
        Ok(())
    }

Get sender of current message. With SEND method, it will be the origin() of the message. With REPORT method, it will be the last element of path.

Examples found in repository?
src/message/protocols/relay.rs (line 143)
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145

    pub fn report(&self) -> Result<Self> {
        if self.method != RelayMethod::SEND {
            return Err(Error::ReportNeedSend);
        }

        if self.path.len() < 2 {
            return Err(Error::CannotInferNextHop);
        }

        Ok(Self {
            method: RelayMethod::REPORT,
            path: self.path.clone(),
            path_end_cursor: 0,
            next_hop: self.path_prev(),
            destination: self.sender(),
        })
    }
More examples
Hide additional examples
src/message/handlers/connection.rs (line 101)
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182

    async fn handle(&self, ctx: &MessagePayload<Message>, msg: &ConnectNodeSend) -> Result<()> {
        let mut relay = ctx.relay.clone();
        // if id is not dest
        if self.dht.did != relay.destination {
            if self
                .swarm
                .get_and_check_transport(relay.destination)
                .await
                .is_some()
            {
                relay.relay(self.dht.did, Some(relay.destination))?;
                return self.forward_payload(ctx, relay).await;
            } else {
                let next_node = match self.dht.find_successor(relay.destination)? {
                    PeerRingAction::Some(node) => Some(node),
                    PeerRingAction::RemoteAction(node, _) => Some(node),
                    _ => None,
                }
                .ok_or(Error::MessageHandlerMissNextNode)?;
                relay.relay(self.dht.did, Some(next_node))?;
                return self.forward_payload(ctx, relay).await;
            }
        } else {
            // self is dest
            relay.relay(self.dht.did, None)?;
            match self.swarm.get_and_check_transport(relay.sender()).await {
                None => {
                    let trans = self.swarm.new_transport().await?;
                    trans
                        .register_remote_info(msg.handshake_info.to_owned().into())
                        .await?;
                    let handshake_info = trans
                        .get_handshake_info(
                            self.swarm.session_manager(),
                            RTCSdpType::Answer,
                            self.swarm.services(),
                        )
                        .await?
                        .to_string();
                    self.send_report_message(
                        Message::ConnectNodeReport(ConnectNodeReport {
                            transport_uuid: msg.transport_uuid.clone(),
                            handshake_info,
                        }),
                        ctx.tx_id,
                        relay,
                    )
                    .await?;
                    self.swarm.push_pending_transport(&trans)?;
                    Ok(())
                }

                _ => {
                    self.send_report_message(
                        Message::AlreadyConnected(AlreadyConnected),
                        ctx.tx_id,
                        relay,
                    )
                    .await
                }
            }
        }
    }
}

#[cfg_attr(feature = "wasm", async_trait(?Send))]
#[cfg_attr(not(feature = "wasm"), async_trait)]
impl HandleMsg<ConnectNodeReport> for MessageHandler {
    async fn handle(&self, ctx: &MessagePayload<Message>, msg: &ConnectNodeReport) -> Result<()> {
        let mut relay = ctx.relay.clone();

        relay.relay(self.dht.did, None)?;
        if relay.next_hop.is_some() {
            self.forward_payload(ctx, relay).await
        } else {
            let transport = self
                .swarm
                .find_pending_transport(
                    uuid::Uuid::from_str(&msg.transport_uuid)
                        .map_err(|_| Error::InvalidTransportUuid)?,
                )?
                .ok_or(Error::MessageHandlerMissTransportConnectedNode)?;
            transport
                .register_remote_info(msg.handshake_info.clone().into())
                .await?;
            Ok(())
        }
    }
}

#[cfg_attr(feature = "wasm", async_trait(?Send))]
#[cfg_attr(not(feature = "wasm"), async_trait)]
impl HandleMsg<AlreadyConnected> for MessageHandler {
    async fn handle(&self, ctx: &MessagePayload<Message>, _msg: &AlreadyConnected) -> Result<()> {
        let mut relay = ctx.relay.clone();

        relay.relay(self.dht.did, None)?;
        if relay.next_hop.is_some() {
            self.forward_payload(ctx, relay).await
        } else {
            self.swarm
                .get_and_check_transport(relay.sender())
                .await
                .map(|_| ())
                .ok_or(Error::MessageHandlerMissTransportAlreadyConnected)
        }
    }

Get the previous element of the element pointed by path_end_cursor.

Examples found in repository?
src/message/protocols/relay.rs (line 119)
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145

    pub fn relay(&mut self, current: Did, next_hop: Option<Did>) -> Result<()> {
        self.validate()?;

        // If self.next_hop is set, it should be current
        if self.next_hop.is_some() && self.next_hop.unwrap() != current {
            return Err(Error::InvalidNextHop);
        }

        match self.method {
            RelayMethod::SEND => {
                self.path.push(current);
                self.next_hop = next_hop;
                Ok(())
            }

            RelayMethod::REPORT => {
                // The final hop
                if self.next_hop == Some(self.destination) {
                    self.path_end_cursor = self.path.len() - 1;
                    self.next_hop = None;
                    return Ok(());
                }

                let pos = self
                    .path
                    .iter()
                    .rev()
                    .skip(self.path_end_cursor)
                    .position(|&x| x == current);

                if let (None, None) = (pos, next_hop) {
                    return Err(Error::CannotInferNextHop);
                }

                if let Some(pos) = pos {
                    self.path_end_cursor += pos;
                }

                // `self.path_prev()` should never return None here, because we have handled final hop
                self.next_hop = next_hop.or_else(|| self.path_prev());

                Ok(())
            }
        }
    }

    /// Construct a `MessageRelay` of method REPORT from a `MessageRelay` of method REPORT.
    /// It will return Error if method is not SEND.
    /// It will return Error if `self.path.len()` is less than 2.
    pub fn report(&self) -> Result<Self> {
        if self.method != RelayMethod::SEND {
            return Err(Error::ReportNeedSend);
        }

        if self.path.len() < 2 {
            return Err(Error::CannotInferNextHop);
        }

        Ok(Self {
            method: RelayMethod::REPORT,
            path: self.path.clone(),
            path_end_cursor: 0,
            next_hop: self.path_prev(),
            destination: self.sender(),
        })
    }

Trait Implementations§

Returns a copy of the value. Read more
Performs copy-assignment from source. Read more
Formats the value using the given formatter. Read more
Deserialize this value from the given Serde deserializer. Read more
This method tests for self and other values to be equal, and is used by ==.
This method tests for !=. The default implementation is almost always sufficient, and should not be overridden without very good reason.
Serialize this value into the given Serde serializer. Read more

Auto Trait Implementations§

Blanket Implementations§

Gets the TypeId of self. Read more
Immutably borrows from an owned value. Read more
Mutably borrows from an owned value. Read more
Compare self to key and return true if they are equal.

Returns the argument unchanged.

Instruments this type with the provided Span, returning an Instrumented wrapper. Read more
Instruments this type with the current Span, returning an Instrumented wrapper. Read more

Calls U::from(self).

That is, this conversion is whatever the implementation of From<T> for U chooses to do.

The alignment of pointer.
The type for initializers.
Initializes a with the given initializer. Read more
Dereferences the given pointer. Read more
Mutably dereferences the given pointer. Read more
Drops the object pointed to by the given pointer. Read more
Should always be Self
The resulting type after obtaining ownership.
Creates owned data from borrowed data, usually by cloning. Read more
Uses borrowed data to replace owned data, usually by cloning. Read more
The type returned in the event of a conversion error.
Performs the conversion.
The type returned in the event of a conversion error.
Performs the conversion.
Attaches the provided Subscriber to this type, returning a WithDispatch wrapper. Read more
Attaches the current default Subscriber to this type, returning a WithDispatch wrapper. Read more