1#![allow(non_snake_case)]
2
3use super::{entity_messages, messages, request_messages, ConnectionId, TypedEnvelope};
4use anyhow::{anyhow, Result};
5use async_tungstenite::tungstenite::Message as WebSocketMessage;
6use collections::HashMap;
7use futures::{SinkExt as _, StreamExt as _};
8use prost::Message as _;
9use serde::Serialize;
10use std::any::{Any, TypeId};
11use std::{
12 cmp,
13 fmt::Debug,
14 io, iter,
15 time::{Duration, SystemTime, UNIX_EPOCH},
16};
17use std::{fmt, mem};
18
19include!(concat!(env!("OUT_DIR"), "/zed.messages.rs"));
20
21pub trait EnvelopedMessage: Clone + Debug + Serialize + Sized + Send + Sync + 'static {
22 const NAME: &'static str;
23 const PRIORITY: MessagePriority;
24 fn into_envelope(
25 self,
26 id: u32,
27 responding_to: Option<u32>,
28 original_sender_id: Option<PeerId>,
29 ) -> Envelope;
30 fn from_envelope(envelope: Envelope) -> Option<Self>;
31}
32
33pub trait EntityMessage: EnvelopedMessage {
34 type Entity;
35 fn remote_entity_id(&self) -> u64;
36}
37
38pub trait RequestMessage: EnvelopedMessage {
39 type Response: EnvelopedMessage;
40}
41
42pub trait AnyTypedEnvelope: 'static + Send + Sync {
43 fn payload_type_id(&self) -> TypeId;
44 fn payload_type_name(&self) -> &'static str;
45 fn as_any(&self) -> &dyn Any;
46 fn into_any(self: Box<Self>) -> Box<dyn Any + Send + Sync>;
47 fn is_background(&self) -> bool;
48 fn original_sender_id(&self) -> Option<PeerId>;
49 fn sender_id(&self) -> ConnectionId;
50 fn message_id(&self) -> u32;
51}
52
53pub enum MessagePriority {
54 Foreground,
55 Background,
56}
57
58impl<T: EnvelopedMessage> AnyTypedEnvelope for TypedEnvelope<T> {
59 fn payload_type_id(&self) -> TypeId {
60 TypeId::of::<T>()
61 }
62
63 fn payload_type_name(&self) -> &'static str {
64 T::NAME
65 }
66
67 fn as_any(&self) -> &dyn Any {
68 self
69 }
70
71 fn into_any(self: Box<Self>) -> Box<dyn Any + Send + Sync> {
72 self
73 }
74
75 fn is_background(&self) -> bool {
76 matches!(T::PRIORITY, MessagePriority::Background)
77 }
78
79 fn original_sender_id(&self) -> Option<PeerId> {
80 self.original_sender_id
81 }
82
83 fn sender_id(&self) -> ConnectionId {
84 self.sender_id
85 }
86
87 fn message_id(&self) -> u32 {
88 self.message_id
89 }
90}
91
92impl PeerId {
93 pub fn from_u64(peer_id: u64) -> Self {
94 let owner_id = (peer_id >> 32) as u32;
95 let id = peer_id as u32;
96 Self { owner_id, id }
97 }
98
99 pub fn as_u64(self) -> u64 {
100 ((self.owner_id as u64) << 32) | (self.id as u64)
101 }
102}
103
104impl Copy for PeerId {}
105
106impl Eq for PeerId {}
107
108impl Ord for PeerId {
109 fn cmp(&self, other: &Self) -> cmp::Ordering {
110 self.owner_id
111 .cmp(&other.owner_id)
112 .then_with(|| self.id.cmp(&other.id))
113 }
114}
115
116impl PartialOrd for PeerId {
117 fn partial_cmp(&self, other: &Self) -> Option<cmp::Ordering> {
118 Some(self.cmp(other))
119 }
120}
121
122impl std::hash::Hash for PeerId {
123 fn hash<H: std::hash::Hasher>(&self, state: &mut H) {
124 self.owner_id.hash(state);
125 self.id.hash(state);
126 }
127}
128
129impl fmt::Display for PeerId {
130 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
131 write!(f, "{}/{}", self.owner_id, self.id)
132 }
133}
134
135messages!(
136 (Ack, Foreground),
137 (AckBufferOperation, Background),
138 (AckChannelMessage, Background),
139 (AddNotification, Foreground),
140 (AddProjectCollaborator, Foreground),
141 (ApplyCodeAction, Background),
142 (ApplyCodeActionResponse, Background),
143 (ApplyCompletionAdditionalEdits, Background),
144 (ApplyCompletionAdditionalEditsResponse, Background),
145 (BufferReloaded, Foreground),
146 (BufferSaved, Foreground),
147 (Call, Foreground),
148 (CallCanceled, Foreground),
149 (CancelCall, Foreground),
150 (ChannelMessageSent, Foreground),
151 (CopyProjectEntry, Foreground),
152 (CreateBufferForPeer, Foreground),
153 (CreateChannel, Foreground),
154 (CreateChannelResponse, Foreground),
155 (CreateProjectEntry, Foreground),
156 (CreateRoom, Foreground),
157 (CreateRoomResponse, Foreground),
158 (DeclineCall, Foreground),
159 (DeleteChannel, Foreground),
160 (DeleteNotification, Foreground),
161 (DeleteProjectEntry, Foreground),
162 (Error, Foreground),
163 (ExpandProjectEntry, Foreground),
164 (ExpandProjectEntryResponse, Foreground),
165 (Follow, Foreground),
166 (FollowResponse, Foreground),
167 (FormatBuffers, Foreground),
168 (FormatBuffersResponse, Foreground),
169 (FuzzySearchUsers, Foreground),
170 (GetChannelMembers, Foreground),
171 (GetChannelMembersResponse, Foreground),
172 (GetChannelMessages, Background),
173 (GetChannelMessagesById, Background),
174 (GetChannelMessagesResponse, Background),
175 (GetCodeActions, Background),
176 (GetCodeActionsResponse, Background),
177 (GetCompletions, Background),
178 (GetCompletionsResponse, Background),
179 (GetDefinition, Background),
180 (GetDefinitionResponse, Background),
181 (GetDocumentHighlights, Background),
182 (GetDocumentHighlightsResponse, Background),
183 (GetHover, Background),
184 (GetHoverResponse, Background),
185 (GetNotifications, Foreground),
186 (GetNotificationsResponse, Foreground),
187 (GetPrivateUserInfo, Foreground),
188 (GetPrivateUserInfoResponse, Foreground),
189 (GetProjectSymbols, Background),
190 (GetProjectSymbolsResponse, Background),
191 (GetReferences, Background),
192 (GetReferencesResponse, Background),
193 (GetTypeDefinition, Background),
194 (GetTypeDefinitionResponse, Background),
195 (GetUsers, Foreground),
196 (Hello, Foreground),
197 (IncomingCall, Foreground),
198 (InlayHints, Background),
199 (InlayHintsResponse, Background),
200 (InviteChannelMember, Foreground),
201 (JoinChannel, Foreground),
202 (JoinChannelBuffer, Foreground),
203 (JoinChannelBufferResponse, Foreground),
204 (JoinChannelChat, Foreground),
205 (JoinChannelChatResponse, Foreground),
206 (JoinProject, Foreground),
207 (JoinProjectResponse, Foreground),
208 (JoinRoom, Foreground),
209 (JoinRoomResponse, Foreground),
210 (LeaveChannelBuffer, Background),
211 (LeaveChannelChat, Foreground),
212 (LeaveProject, Foreground),
213 (LeaveRoom, Foreground),
214 (MarkNotificationRead, Foreground),
215 (MoveChannel, Foreground),
216 (OnTypeFormatting, Background),
217 (OnTypeFormattingResponse, Background),
218 (OpenBufferById, Background),
219 (OpenBufferByPath, Background),
220 (OpenBufferForSymbol, Background),
221 (OpenBufferForSymbolResponse, Background),
222 (OpenBufferResponse, Background),
223 (PerformRename, Background),
224 (PerformRenameResponse, Background),
225 (Ping, Foreground),
226 (PrepareRename, Background),
227 (PrepareRenameResponse, Background),
228 (ProjectEntryResponse, Foreground),
229 (RefreshInlayHints, Foreground),
230 (RejoinChannelBuffers, Foreground),
231 (RejoinChannelBuffersResponse, Foreground),
232 (RejoinRoom, Foreground),
233 (RejoinRoomResponse, Foreground),
234 (ReloadBuffers, Foreground),
235 (ReloadBuffersResponse, Foreground),
236 (RemoveChannelMember, Foreground),
237 (RemoveChannelMessage, Foreground),
238 (RemoveContact, Foreground),
239 (RemoveProjectCollaborator, Foreground),
240 (RenameChannel, Foreground),
241 (RenameChannelResponse, Foreground),
242 (RenameProjectEntry, Foreground),
243 (RequestContact, Foreground),
244 (ResolveCompletionDocumentation, Background),
245 (ResolveCompletionDocumentationResponse, Background),
246 (ResolveInlayHint, Background),
247 (ResolveInlayHintResponse, Background),
248 (RespondToChannelInvite, Foreground),
249 (RespondToContactRequest, Foreground),
250 (RoomUpdated, Foreground),
251 (SaveBuffer, Foreground),
252 (SetChannelMemberRole, Foreground),
253 (SetChannelVisibility, Foreground),
254 (SearchProject, Background),
255 (SearchProjectResponse, Background),
256 (SendChannelMessage, Background),
257 (SendChannelMessageResponse, Background),
258 (ShareProject, Foreground),
259 (ShareProjectResponse, Foreground),
260 (ShowContacts, Foreground),
261 (StartLanguageServer, Foreground),
262 (SynchronizeBuffers, Foreground),
263 (SynchronizeBuffersResponse, Foreground),
264 (Test, Foreground),
265 (Unfollow, Foreground),
266 (UnshareProject, Foreground),
267 (UpdateBuffer, Foreground),
268 (UpdateBufferFile, Foreground),
269 (UpdateChannelBuffer, Foreground),
270 (UpdateChannelBufferCollaborators, Foreground),
271 (UpdateChannels, Foreground),
272 (UpdateUserChannels, Foreground),
273 (UpdateContacts, Foreground),
274 (UpdateDiagnosticSummary, Foreground),
275 (UpdateDiffBase, Foreground),
276 (UpdateFollowers, Foreground),
277 (UpdateInviteInfo, Foreground),
278 (UpdateLanguageServer, Foreground),
279 (UpdateParticipantLocation, Foreground),
280 (UpdateProject, Foreground),
281 (UpdateProjectCollaborator, Foreground),
282 (UpdateWorktree, Foreground),
283 (UpdateWorktreeSettings, Foreground),
284 (UsersResponse, Foreground),
285 (LspExtExpandMacro, Background),
286 (LspExtExpandMacroResponse, Background),
287 (SetRoomParticipantRole, Foreground),
288);
289
290request_messages!(
291 (ApplyCodeAction, ApplyCodeActionResponse),
292 (
293 ApplyCompletionAdditionalEdits,
294 ApplyCompletionAdditionalEditsResponse
295 ),
296 (Call, Ack),
297 (CancelCall, Ack),
298 (CopyProjectEntry, ProjectEntryResponse),
299 (CreateChannel, CreateChannelResponse),
300 (CreateProjectEntry, ProjectEntryResponse),
301 (CreateRoom, CreateRoomResponse),
302 (DeclineCall, Ack),
303 (DeleteChannel, Ack),
304 (DeleteProjectEntry, ProjectEntryResponse),
305 (ExpandProjectEntry, ExpandProjectEntryResponse),
306 (Follow, FollowResponse),
307 (FormatBuffers, FormatBuffersResponse),
308 (FuzzySearchUsers, UsersResponse),
309 (GetChannelMembers, GetChannelMembersResponse),
310 (GetChannelMessages, GetChannelMessagesResponse),
311 (GetChannelMessagesById, GetChannelMessagesResponse),
312 (GetCodeActions, GetCodeActionsResponse),
313 (GetCompletions, GetCompletionsResponse),
314 (GetDefinition, GetDefinitionResponse),
315 (GetDocumentHighlights, GetDocumentHighlightsResponse),
316 (GetHover, GetHoverResponse),
317 (GetNotifications, GetNotificationsResponse),
318 (GetPrivateUserInfo, GetPrivateUserInfoResponse),
319 (GetProjectSymbols, GetProjectSymbolsResponse),
320 (GetReferences, GetReferencesResponse),
321 (GetTypeDefinition, GetTypeDefinitionResponse),
322 (GetUsers, UsersResponse),
323 (IncomingCall, Ack),
324 (InlayHints, InlayHintsResponse),
325 (InviteChannelMember, Ack),
326 (JoinChannel, JoinRoomResponse),
327 (JoinChannelBuffer, JoinChannelBufferResponse),
328 (JoinChannelChat, JoinChannelChatResponse),
329 (JoinProject, JoinProjectResponse),
330 (JoinRoom, JoinRoomResponse),
331 (LeaveChannelBuffer, Ack),
332 (LeaveRoom, Ack),
333 (MarkNotificationRead, Ack),
334 (MoveChannel, Ack),
335 (OnTypeFormatting, OnTypeFormattingResponse),
336 (OpenBufferById, OpenBufferResponse),
337 (OpenBufferByPath, OpenBufferResponse),
338 (OpenBufferForSymbol, OpenBufferForSymbolResponse),
339 (PerformRename, PerformRenameResponse),
340 (Ping, Ack),
341 (PrepareRename, PrepareRenameResponse),
342 (RefreshInlayHints, Ack),
343 (RejoinChannelBuffers, RejoinChannelBuffersResponse),
344 (RejoinRoom, RejoinRoomResponse),
345 (ReloadBuffers, ReloadBuffersResponse),
346 (RemoveChannelMember, Ack),
347 (RemoveChannelMessage, Ack),
348 (RemoveContact, Ack),
349 (RenameChannel, RenameChannelResponse),
350 (RenameProjectEntry, ProjectEntryResponse),
351 (RequestContact, Ack),
352 (
353 ResolveCompletionDocumentation,
354 ResolveCompletionDocumentationResponse
355 ),
356 (ResolveInlayHint, ResolveInlayHintResponse),
357 (RespondToChannelInvite, Ack),
358 (RespondToContactRequest, Ack),
359 (SaveBuffer, BufferSaved),
360 (SearchProject, SearchProjectResponse),
361 (SendChannelMessage, SendChannelMessageResponse),
362 (SetChannelMemberRole, Ack),
363 (SetChannelVisibility, Ack),
364 (ShareProject, ShareProjectResponse),
365 (SynchronizeBuffers, SynchronizeBuffersResponse),
366 (Test, Test),
367 (UpdateBuffer, Ack),
368 (UpdateParticipantLocation, Ack),
369 (UpdateProject, Ack),
370 (UpdateWorktree, Ack),
371 (LspExtExpandMacro, LspExtExpandMacroResponse),
372 (SetRoomParticipantRole, Ack),
373);
374
375entity_messages!(
376 {project_id, ShareProject},
377 AddProjectCollaborator,
378 ApplyCodeAction,
379 ApplyCompletionAdditionalEdits,
380 BufferReloaded,
381 BufferSaved,
382 CopyProjectEntry,
383 CreateBufferForPeer,
384 CreateProjectEntry,
385 DeleteProjectEntry,
386 ExpandProjectEntry,
387 FormatBuffers,
388 GetCodeActions,
389 GetCompletions,
390 GetDefinition,
391 GetDocumentHighlights,
392 GetHover,
393 GetProjectSymbols,
394 GetReferences,
395 GetTypeDefinition,
396 InlayHints,
397 JoinProject,
398 LeaveProject,
399 OnTypeFormatting,
400 OpenBufferById,
401 OpenBufferByPath,
402 OpenBufferForSymbol,
403 PerformRename,
404 PrepareRename,
405 RefreshInlayHints,
406 ReloadBuffers,
407 RemoveProjectCollaborator,
408 RenameProjectEntry,
409 ResolveCompletionDocumentation,
410 ResolveInlayHint,
411 SaveBuffer,
412 SearchProject,
413 StartLanguageServer,
414 SynchronizeBuffers,
415 UnshareProject,
416 UpdateBuffer,
417 UpdateBufferFile,
418 UpdateDiagnosticSummary,
419 UpdateDiffBase,
420 UpdateLanguageServer,
421 UpdateProject,
422 UpdateProjectCollaborator,
423 UpdateWorktree,
424 UpdateWorktreeSettings,
425 LspExtExpandMacro,
426);
427
428entity_messages!(
429 {channel_id, Channel},
430 ChannelMessageSent,
431 RemoveChannelMessage,
432 UpdateChannelBuffer,
433 UpdateChannelBufferCollaborators,
434);
435
436const KIB: usize = 1024;
437const MIB: usize = KIB * 1024;
438const MAX_BUFFER_LEN: usize = MIB;
439
440/// A stream of protobuf messages.
441pub struct MessageStream<S> {
442 stream: S,
443 encoding_buffer: Vec<u8>,
444}
445
446#[allow(clippy::large_enum_variant)]
447#[derive(Debug)]
448pub enum Message {
449 Envelope(Envelope),
450 Ping,
451 Pong,
452}
453
454impl<S> MessageStream<S> {
455 pub fn new(stream: S) -> Self {
456 Self {
457 stream,
458 encoding_buffer: Vec::new(),
459 }
460 }
461
462 pub fn inner_mut(&mut self) -> &mut S {
463 &mut self.stream
464 }
465}
466
467impl<S> MessageStream<S>
468where
469 S: futures::Sink<WebSocketMessage, Error = anyhow::Error> + Unpin,
470{
471 pub async fn write(&mut self, message: Message) -> Result<(), anyhow::Error> {
472 #[cfg(any(test, feature = "test-support"))]
473 const COMPRESSION_LEVEL: i32 = -7;
474
475 #[cfg(not(any(test, feature = "test-support")))]
476 const COMPRESSION_LEVEL: i32 = 4;
477
478 match message {
479 Message::Envelope(message) => {
480 self.encoding_buffer.reserve(message.encoded_len());
481 message
482 .encode(&mut self.encoding_buffer)
483 .map_err(io::Error::from)?;
484 let buffer =
485 zstd::stream::encode_all(self.encoding_buffer.as_slice(), COMPRESSION_LEVEL)
486 .unwrap();
487
488 self.encoding_buffer.clear();
489 self.encoding_buffer.shrink_to(MAX_BUFFER_LEN);
490 self.stream.send(WebSocketMessage::Binary(buffer)).await?;
491 }
492 Message::Ping => {
493 self.stream
494 .send(WebSocketMessage::Ping(Default::default()))
495 .await?;
496 }
497 Message::Pong => {
498 self.stream
499 .send(WebSocketMessage::Pong(Default::default()))
500 .await?;
501 }
502 }
503
504 Ok(())
505 }
506}
507
508impl<S> MessageStream<S>
509where
510 S: futures::Stream<Item = Result<WebSocketMessage, anyhow::Error>> + Unpin,
511{
512 pub async fn read(&mut self) -> Result<Message, anyhow::Error> {
513 while let Some(bytes) = self.stream.next().await {
514 match bytes? {
515 WebSocketMessage::Binary(bytes) => {
516 zstd::stream::copy_decode(bytes.as_slice(), &mut self.encoding_buffer).unwrap();
517 let envelope = Envelope::decode(self.encoding_buffer.as_slice())
518 .map_err(io::Error::from)?;
519
520 self.encoding_buffer.clear();
521 self.encoding_buffer.shrink_to(MAX_BUFFER_LEN);
522 return Ok(Message::Envelope(envelope));
523 }
524 WebSocketMessage::Ping(_) => return Ok(Message::Ping),
525 WebSocketMessage::Pong(_) => return Ok(Message::Pong),
526 WebSocketMessage::Close(_) => break,
527 _ => {}
528 }
529 }
530 Err(anyhow!("connection closed"))
531 }
532}
533
534impl From<Timestamp> for SystemTime {
535 fn from(val: Timestamp) -> Self {
536 UNIX_EPOCH
537 .checked_add(Duration::new(val.seconds, val.nanos))
538 .unwrap()
539 }
540}
541
542impl From<SystemTime> for Timestamp {
543 fn from(time: SystemTime) -> Self {
544 let duration = time.duration_since(UNIX_EPOCH).unwrap();
545 Self {
546 seconds: duration.as_secs(),
547 nanos: duration.subsec_nanos(),
548 }
549 }
550}
551
552impl From<u128> for Nonce {
553 fn from(nonce: u128) -> Self {
554 let upper_half = (nonce >> 64) as u64;
555 let lower_half = nonce as u64;
556 Self {
557 upper_half,
558 lower_half,
559 }
560 }
561}
562
563impl From<Nonce> for u128 {
564 fn from(nonce: Nonce) -> Self {
565 let upper_half = (nonce.upper_half as u128) << 64;
566 let lower_half = nonce.lower_half as u128;
567 upper_half | lower_half
568 }
569}
570
571pub fn split_worktree_update(
572 mut message: UpdateWorktree,
573 max_chunk_size: usize,
574) -> impl Iterator<Item = UpdateWorktree> {
575 let mut done_files = false;
576
577 let mut repository_map = message
578 .updated_repositories
579 .into_iter()
580 .map(|repo| (repo.work_directory_id, repo))
581 .collect::<HashMap<_, _>>();
582
583 iter::from_fn(move || {
584 if done_files {
585 return None;
586 }
587
588 let updated_entries_chunk_size = cmp::min(message.updated_entries.len(), max_chunk_size);
589 let updated_entries: Vec<_> = message
590 .updated_entries
591 .drain(..updated_entries_chunk_size)
592 .collect();
593
594 let removed_entries_chunk_size = cmp::min(message.removed_entries.len(), max_chunk_size);
595 let removed_entries = message
596 .removed_entries
597 .drain(..removed_entries_chunk_size)
598 .collect();
599
600 done_files = message.updated_entries.is_empty() && message.removed_entries.is_empty();
601
602 let mut updated_repositories = Vec::new();
603
604 if !repository_map.is_empty() {
605 for entry in &updated_entries {
606 if let Some(repo) = repository_map.remove(&entry.id) {
607 updated_repositories.push(repo)
608 }
609 }
610 }
611
612 let removed_repositories = if done_files {
613 mem::take(&mut message.removed_repositories)
614 } else {
615 Default::default()
616 };
617
618 if done_files {
619 updated_repositories.extend(mem::take(&mut repository_map).into_values());
620 }
621
622 Some(UpdateWorktree {
623 project_id: message.project_id,
624 worktree_id: message.worktree_id,
625 root_name: message.root_name.clone(),
626 abs_path: message.abs_path.clone(),
627 updated_entries,
628 removed_entries,
629 scan_id: message.scan_id,
630 is_last_update: done_files && message.is_last_update,
631 updated_repositories,
632 removed_repositories,
633 })
634 })
635}
636
637#[cfg(test)]
638mod tests {
639 use super::*;
640
641 #[gpui::test]
642 async fn test_buffer_size() {
643 let (tx, rx) = futures::channel::mpsc::unbounded();
644 let mut sink = MessageStream::new(tx.sink_map_err(|_| anyhow!("")));
645 sink.write(Message::Envelope(Envelope {
646 payload: Some(envelope::Payload::UpdateWorktree(UpdateWorktree {
647 root_name: "abcdefg".repeat(10),
648 ..Default::default()
649 })),
650 ..Default::default()
651 }))
652 .await
653 .unwrap();
654 assert!(sink.encoding_buffer.capacity() <= MAX_BUFFER_LEN);
655 sink.write(Message::Envelope(Envelope {
656 payload: Some(envelope::Payload::UpdateWorktree(UpdateWorktree {
657 root_name: "abcdefg".repeat(1000000),
658 ..Default::default()
659 })),
660 ..Default::default()
661 }))
662 .await
663 .unwrap();
664 assert!(sink.encoding_buffer.capacity() <= MAX_BUFFER_LEN);
665
666 let mut stream = MessageStream::new(rx.map(anyhow::Ok));
667 stream.read().await.unwrap();
668 assert!(stream.encoding_buffer.capacity() <= MAX_BUFFER_LEN);
669 stream.read().await.unwrap();
670 assert!(stream.encoding_buffer.capacity() <= MAX_BUFFER_LEN);
671 }
672
673 #[gpui::test]
674 fn test_converting_peer_id_from_and_to_u64() {
675 let peer_id = PeerId {
676 owner_id: 10,
677 id: 3,
678 };
679 assert_eq!(PeerId::from_u64(peer_id.as_u64()), peer_id);
680 let peer_id = PeerId {
681 owner_id: u32::MAX,
682 id: 3,
683 };
684 assert_eq!(PeerId::from_u64(peer_id.as_u64()), peer_id);
685 let peer_id = PeerId {
686 owner_id: 10,
687 id: u32::MAX,
688 };
689 assert_eq!(PeerId::from_u64(peer_id.as_u64()), peer_id);
690 let peer_id = PeerId {
691 owner_id: u32::MAX,
692 id: u32::MAX,
693 };
694 assert_eq!(PeerId::from_u64(peer_id.as_u64()), peer_id);
695 }
696}