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
|
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),
}
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,
}
}
}
#[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,
}
/// 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>;
}
|
