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, 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> + Send>> + Send + Sync + 'static; pub type MessageHandler = Box; #[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; fn encode(&self, peer: &str, message: &ParsedMessage) -> Result; 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; }