1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
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
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
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
|
use anyhow::Result;
use async_trait::async_trait;
use automerge::sync::Message;
use std::boxed::Box;
use std::future::Future;
use std::pin::Pin;
pub use crate::replica::Epoch;
pub type HeadsDigest = [u8; 32];
#[derive(Debug, Clone)]
pub struct Keepalive {
pub heads_digest: HeadsDigest,
/// The digest is only comparable between nodes on the same epoch.
pub epoch: Epoch,
}
#[derive(Debug)]
pub enum ParsedMessage {
Hello(HelloRequest),
HelloReply(HelloReply),
Sync(SyncEnvelope),
Ping(Keepalive),
Pong(Keepalive),
/// Asks a peer for its document of a newer epoch (late or new nodes).
SnapshotRequest(SnapshotRequest),
SnapshotReply(SnapshotReply),
}
impl ParsedMessage {
/// Every message carries the sender's epoch.
pub fn epoch(&self) -> Epoch {
match self {
ParsedMessage::Hello(hello) => hello.epoch,
ParsedMessage::HelloReply(reply) => reply.epoch,
ParsedMessage::Sync(envelope) => envelope.epoch,
ParsedMessage::Ping(keepalive) | ParsedMessage::Pong(keepalive) => keepalive.epoch,
ParsedMessage::SnapshotRequest(request) => request.epoch,
ParsedMessage::SnapshotReply(reply) => reply.epoch,
}
}
}
#[derive(Debug, Clone)]
pub struct HelloRequest {
pub epoch: Epoch,
}
#[derive(Debug, Clone)]
pub struct HelloReply {
pub cluster_id: String,
pub peers: Vec<String>,
pub epoch: Epoch,
}
#[derive(Debug, Clone)]
pub struct SyncEnvelope {
pub cluster_id: String,
/// Epoch of the document the message was generated from.
pub epoch: Epoch,
pub message: Message,
}
#[derive(Debug, Clone)]
pub struct SnapshotRequest {
/// The requester's epoch.
pub epoch: Epoch,
/// The epoch whose document is wanted.
pub wanted: Epoch,
}
#[derive(Clone)]
pub struct SnapshotReply {
/// The sender's epoch, which is the epoch of `document`.
pub epoch: Epoch,
/// `AutoCommit::save()` of the sender's document; empty if it isn't on the wanted
/// epoch (any more).
pub document: Vec<u8>,
}
impl std::fmt::Debug for SnapshotReply {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("SnapshotReply")
.field("epoch", &self.epoch)
.field("document_bytes", &self.document.len())
.finish()
}
}
/// A message serialized for the wire. Cloning is cheap (shared buffer).
#[derive(Clone, Debug)]
pub struct EncodedMessage {
pub peer: String,
pub body: bytes::Bytes,
}
pub type HandlerFn = dyn Fn(String, ParsedMessage) -> Pin<Box<dyn Future<Output = Result<()>> + Send>>
+ Send
+ Sync
+ 'static;
pub type MessageHandler = Box<HandlerFn>;
#[async_trait]
pub trait Transport: Send + Sync {
/// This node's identity on the transport (for Mycelium: its public key).
fn local_id(&self) -> &str;
/// Resolves a user-supplied peer hint (e.g. a seed) to a peer identity.
async fn resolve_peer(&self, hint: &str) -> Result<String>;
fn encode(&self, peer: &str, message: &ParsedMessage) -> Result<EncodedMessage>;
async fn send_encoded(&self, message: &EncodedMessage) -> Result<()>;
/// Sends an outbound message payload to a targeted peer on a specific topic.
async fn send_message(&self, peer: &str, message: &ParsedMessage) -> Result<()> {
let encoded = self.encode(peer, message)?;
self.send_encoded(&encoded).await
}
/// Starts the long-polling inbound event loops, driving messages to the provided handler closure.
async fn start_consumer(&self, handler: MessageHandler) -> Result<()>;
/// Asynchronously monitors for fatal network client or sub-process failures.
async fn wait_for_exit(&self) -> std::io::Result<std::process::ExitStatus>;
}
|
