summaryrefslogtreecommitdiffstats
path: root/src
diff options
context:
space:
mode:
Diffstat (limited to 'src')
-rw-r--r--src/api/auth.rs6
-rw-r--r--src/document/mod.rs4
-rw-r--r--src/document/replica.rs38
-rw-r--r--src/document/store.rs41
-rw-r--r--src/transport/firewall.rs20
-rw-r--r--src/transport/mod.rs128
-rw-r--r--src/transport/mycelium/daemon.rs50
-rw-r--r--src/transport/mycelium/mod.rs14
-rw-r--r--src/transport/traffic.rs40
-rw-r--r--src/transport/wire.rs2
10 files changed, 175 insertions, 168 deletions
diff --git a/src/api/auth.rs b/src/api/auth.rs
index c07db74..d58640e 100644
--- a/src/api/auth.rs
+++ b/src/api/auth.rs
@@ -50,7 +50,7 @@ fn load_users(path: &str) -> anyhow::Result<HashMap<String, String>> {
}
#[derive(Clone)]
-pub struct AuthState {
+pub(crate) struct AuthState {
idle_timeout: Duration,
users: Arc<HashMap<String, String>>, // user -> PHC hash string
@@ -60,7 +60,7 @@ pub struct AuthState {
}
impl AuthState {
- pub fn new(path: &str) -> anyhow::Result<Self> {
+ pub(crate) fn new(path: &str) -> anyhow::Result<Self> {
let users = load_users(path)?;
Ok(Self {
@@ -136,7 +136,7 @@ async fn verify(state: &AuthState, user: &str, pass: String, hash: String) -> bo
verified
}
-pub async fn basic_auth(
+pub(crate) async fn basic_auth(
State(state): State<AuthState>,
headers: HeaderMap,
mut request: Request<Body>,
diff --git a/src/document/mod.rs b/src/document/mod.rs
index 1024859..3f489aa 100644
--- a/src/document/mod.rs
+++ b/src/document/mod.rs
@@ -3,5 +3,5 @@
pub(crate) mod file;
pub(crate) mod meta;
-pub mod replica;
-pub mod store;
+pub(crate) mod replica;
+pub(crate) mod store;
diff --git a/src/document/replica.rs b/src/document/replica.rs
index 0cab0be..fe76c12 100644
--- a/src/document/replica.rs
+++ b/src/document/replica.rs
@@ -20,14 +20,14 @@ use tokio::sync::watch;
/// * `id` is the hash of the epoch's snapshot change. It tells two epochs of the same
/// generation apart, which only happens if two remakes committed concurrently.
#[derive(Clone, Copy, PartialEq, Eq, Hash)]
-pub struct Epoch {
- pub generation: u64,
- pub id: [u8; 32],
+pub(crate) struct Epoch {
+ pub(crate) generation: u64,
+ pub(crate) id: [u8; 32],
}
/// How a peer's epoch relates to ours.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
-pub enum EpochRelation {
+pub(crate) enum EpochRelation {
Same,
/// The peer is on an older generation and has to migrate.
PeerBehind,
@@ -39,12 +39,12 @@ pub enum EpochRelation {
impl Epoch {
/// Swarms that have never been compacted.
- pub const GENESIS: Epoch = Epoch {
+ pub(crate) const GENESIS: Epoch = Epoch {
generation: 0,
id: [0; 32],
};
- pub fn relation_to(&self, peer: &Epoch) -> EpochRelation {
+ pub(crate) fn relation_to(&self, peer: &Epoch) -> EpochRelation {
use std::cmp::Ordering;
match peer.generation.cmp(&self.generation) {
Ordering::Less => EpochRelation::PeerBehind,
@@ -54,7 +54,7 @@ impl Epoch {
}
}
- pub fn id_hex(&self) -> String {
+ pub(crate) fn id_hex(&self) -> String {
self.id.iter().map(|byte| format!("{byte:02x}")).collect()
}
}
@@ -90,12 +90,12 @@ impl fmt::Debug for Epoch {
/// The refresh also publishes a [`Stamp`], for what must not wait for the lock: a
/// tokio read lock waits behind a writer, and applying a large sync message holds the
/// write lock for seconds.
-pub struct Replica {
- pub epoch: Epoch,
+pub(crate) struct Replica {
+ pub(crate) epoch: Epoch,
/// How many times this node changed epoch (switch, rejoin, abandoned rejoin) in this
/// run: messages carry it, so a late one can't flip a peer's view of our epoch back.
- pub epoch_seq: u32,
- pub doc: AutoCommit,
+ pub(crate) epoch_seq: u32,
+ pub(crate) doc: AutoCommit,
heads: Vec<ChangeHash>,
stamp: watch::Sender<Stamp>,
}
@@ -103,15 +103,15 @@ pub struct Replica {
/// The epoch label and the heads of the replica, as of its last refresh, published
/// together: keepalives and HELLOs read them without the document's lock.
#[derive(Clone, Debug)]
-pub struct Stamp {
- pub epoch: Epoch,
- pub epoch_seq: u32,
+pub(crate) struct Stamp {
+ pub(crate) epoch: Epoch,
+ pub(crate) epoch_seq: u32,
/// Sorted.
- pub heads: Arc<[ChangeHash]>,
+ pub(crate) heads: Arc<[ChangeHash]>,
}
impl Replica {
- pub fn new(epoch: Epoch, doc: AutoCommit) -> Self {
+ pub(crate) fn new(epoch: Epoch, doc: AutoCommit) -> Self {
let (stamp, _) = watch::channel(Stamp {
epoch,
epoch_seq: 0,
@@ -129,13 +129,13 @@ impl Replica {
}
/// The document's heads as of the last refresh, sorted.
- pub fn heads(&self) -> &[ChangeHash] {
+ pub(crate) fn heads(&self) -> &[ChangeHash] {
&self.heads
}
/// Call after changing the document or its epoch (under the write lock). Publishes
/// the new [`Stamp`].
- pub fn refresh_heads(&mut self) {
+ pub(crate) fn refresh_heads(&mut self) {
self.heads = self.doc.get_heads();
self.heads.sort_unstable();
self.stamp.send_replace(Stamp {
@@ -146,7 +146,7 @@ impl Replica {
}
/// The published stamps, from the last refresh on.
- pub fn stamps(&self) -> watch::Receiver<Stamp> {
+ pub(crate) fn stamps(&self) -> watch::Receiver<Stamp> {
self.stamp.subscribe()
}
}
diff --git a/src/document/store.rs b/src/document/store.rs
index 573b811..132a2ef 100644
--- a/src/document/store.rs
+++ b/src/document/store.rs
@@ -10,7 +10,7 @@ use serde_json::Value as JsonValue;
use std::collections::BTreeSet;
#[derive(Debug, thiserror::Error)]
-pub enum StoreError {
+pub(crate) enum StoreError {
#[error("path crosses a non-map value")]
Conflict,
#[error("value does not exist")]
@@ -204,11 +204,14 @@ fn read_path_inner(
}
/// Reads a path; the empty path is the whole store.
-pub fn read_path(doc: &AutoCommit, path: &[String]) -> Result<Option<JsonValue>, StoreError> {
+pub(crate) fn read_path(
+ doc: &AutoCommit,
+ path: &[String],
+) -> Result<Option<JsonValue>, StoreError> {
read_path_inner(doc, path, None)
}
-pub fn read_path_with_conflicts(
+pub(crate) fn read_path_with_conflicts(
doc: &AutoCommit,
path: &[String],
) -> Result<Option<(JsonValue, Vec<ConflictReport>)>, StoreError> {
@@ -222,7 +225,7 @@ pub fn read_path_with_conflicts(
/// For streams: the value, plus whether the path or subtree involves any conflict.
/// Reports are collected without values, so this costs little more than read_path.
-pub fn read_path_traced(
+pub(crate) fn read_path_traced(
doc: &AutoCommit,
path: &[String],
) -> Result<(Option<JsonValue>, bool), StoreError> {
@@ -232,7 +235,7 @@ pub fn read_path_traced(
}
/// Keys of the merged map at `path`; empty if it doesn't exist or isn't a map.
-pub fn keys_at(doc: &AutoCommit, path: &[String]) -> Result<Vec<String>, StoreError> {
+pub(crate) fn keys_at(doc: &AutoCommit, path: &[String]) -> Result<Vec<String>, StoreError> {
let mut maps = vec![automerge::ROOT];
for segment in path {
maps = match resolve(doc, &maps, segment, None)? {
@@ -267,7 +270,7 @@ pub(crate) fn read_candidates(
}
/// Writes (replaces) the value at `path`, creating intermediate maps.
-pub fn write_path(
+pub(crate) fn write_path(
doc: &mut AutoCommit,
path: &[String],
value: &JsonValue,
@@ -353,7 +356,11 @@ pub(crate) fn copy_path_to(
}
/// Deletes the value at `path`, optionally pruning ancestors that become empty.
-pub fn delete_path(doc: &mut AutoCommit, path: &[String], prune: bool) -> Result<(), StoreError> {
+pub(crate) fn delete_path(
+ doc: &mut AutoCommit,
+ path: &[String],
+ prune: bool,
+) -> Result<(), StoreError> {
let (leaf, parents) = path.split_last().ok_or(StoreError::NotFound)?;
// levels[d] = merged maps at path[..d]
@@ -615,29 +622,29 @@ fn scalar_to_json(scalar: ScalarValue) -> Result<JsonValue, StoreError> {
#[derive(Debug, Serialize)]
#[serde(rename_all = "camelCase")]
-pub enum Resolution {
+pub(crate) enum Resolution {
MergedMaps,
NewestWins,
}
#[derive(Debug, Serialize)]
#[serde(rename_all = "camelCase")]
-pub struct CandidateReport {
- pub op_id: String,
- pub selected: bool,
+pub(crate) struct CandidateReport {
+ pub(crate) op_id: String,
+ pub(crate) selected: bool,
#[serde(skip_serializing_if = "Option::is_none")]
- pub value: Option<JsonValue>,
+ pub(crate) value: Option<JsonValue>,
}
#[derive(Debug, Serialize)]
#[serde(rename_all = "camelCase")]
-pub struct ConflictReport {
- pub path: Vec<String>,
+pub(crate) struct ConflictReport {
+ pub(crate) path: Vec<String>,
/// On the way to the requested path; reported without candidate values.
#[serde(skip_serializing_if = "std::ops::Not::not")]
- pub ancestor: bool,
- pub resolution: Resolution,
- pub candidates: Vec<CandidateReport>,
+ pub(crate) ancestor: bool,
+ pub(crate) resolution: Resolution,
+ pub(crate) candidates: Vec<CandidateReport>,
}
#[derive(Default)]
diff --git a/src/transport/firewall.rs b/src/transport/firewall.rs
index 037e739..2cc1dba 100644
--- a/src/transport/firewall.rs
+++ b/src/transport/firewall.rs
@@ -10,18 +10,18 @@ use crate::transport::EncodedMessage;
use crate::transport::{MessageHandler, ParsedMessage, Transport};
#[derive(Debug, Clone, Default, Serialize, Deserialize)]
-pub struct FirewallRules {
+pub(crate) struct FirewallRules {
/// Messages from these peers are dropped before the application sees them.
#[serde(default)]
- pub inbound: HashSet<String>,
+ pub(crate) inbound: HashSet<String>,
/// Messages to these peers are silently dropped, like a lossy network.
#[serde(default)]
- pub outbound: HashSet<String>,
+ pub(crate) outbound: HashSet<String>,
}
/// Uses a std lock: checks are tiny and the guard is never held across an await.
#[derive(Default)]
-pub struct Firewall {
+pub(crate) struct Firewall {
rules: RwLock<FirewallRules>,
}
@@ -30,30 +30,30 @@ impl Firewall {
self.rules.read_or_recover()
}
- pub fn rules(&self) -> FirewallRules {
+ pub(crate) fn rules(&self) -> FirewallRules {
self.read().clone()
}
- pub fn set_rules(&self, rules: FirewallRules) {
+ pub(crate) fn set_rules(&self, rules: FirewallRules) {
*self.rules.write_or_recover() = rules;
}
- pub fn blocks_inbound(&self, peer: &str) -> bool {
+ pub(crate) fn blocks_inbound(&self, peer: &str) -> bool {
self.read().inbound.contains(peer)
}
- pub fn blocks_outbound(&self, peer: &str) -> bool {
+ pub(crate) fn blocks_outbound(&self, peer: &str) -> bool {
self.read().outbound.contains(peer)
}
}
-pub struct FirewallTransport {
+pub(crate) struct FirewallTransport {
inner: Arc<dyn Transport>,
firewall: Arc<Firewall>,
}
impl FirewallTransport {
- pub fn new(inner: Arc<dyn Transport>, firewall: Arc<Firewall>) -> Self {
+ pub(crate) fn new(inner: Arc<dyn Transport>, firewall: Arc<Firewall>) -> Self {
Self { inner, firewall }
}
}
diff --git a/src/transport/mod.rs b/src/transport/mod.rs
index 0d761e0..c109b0c 100644
--- a/src/transport/mod.rs
+++ b/src/transport/mod.rs
@@ -9,29 +9,29 @@ use std::future::Future;
use std::pin::Pin;
use std::sync::atomic::{AtomicUsize, Ordering};
-pub use crate::document::replica::Epoch;
+pub(crate) use crate::document::replica::Epoch;
-pub mod firewall;
-pub mod mycelium;
+pub(crate) mod firewall;
+pub(crate) mod mycelium;
#[cfg(test)]
pub(crate) mod sim;
-pub mod traffic;
+pub(crate) mod traffic;
pub(crate) mod wire;
/// Random identifier of one run of a node (from start to exit).
-pub type RunId = [u8; 16];
+pub(crate) type RunId = [u8; 16];
/// Most heads listed in a keepalive; more are sent as a digest. A head takes 34 bytes on
/// the wire, so 32 of them keep a keepalive within one Mycelium chunk (1300 bytes; see
/// the size test in `mycelium`). There are rarely more heads than nodes that wrote
/// concurrently since the last merging write.
-pub const MAX_LISTED_HEADS: usize = 32;
+pub(crate) const MAX_LISTED_HEADS: usize = 32;
/// A document's heads, compactly: the heads themselves while there are few, otherwise a
/// digest of them with some of them. Two summaries are equal exactly when the heads are,
/// whatever form each took (nodes with different limits still agree).
#[derive(Debug, Clone)]
-pub enum HeadsSummary {
+pub(crate) enum HeadsSummary {
/// Sorted.
Heads(Vec<ChangeHash>),
Digest {
@@ -48,7 +48,7 @@ pub enum HeadsSummary {
static SAMPLE_OFFSET: AtomicUsize = AtomicUsize::new(0);
impl HeadsSummary {
- pub fn of(mut heads: Vec<ChangeHash>) -> Self {
+ pub(crate) fn of(mut heads: Vec<ChangeHash>) -> Self {
heads.sort_unstable();
if heads.len() <= MAX_LISTED_HEADS {
return Self::Heads(heads);
@@ -68,7 +68,7 @@ impl HeadsSummary {
}
/// The digest of the heads, computed if they are listed.
- pub fn digest(&self) -> [u8; 32] {
+ pub(crate) fn digest(&self) -> [u8; 32] {
match self {
Self::Heads(heads) => digest_of(heads),
Self::Digest { digest, .. } => *digest,
@@ -76,7 +76,7 @@ impl HeadsSummary {
}
/// The heads, if all are listed.
- pub fn heads(&self) -> Option<&[ChangeHash]> {
+ pub(crate) fn heads(&self) -> Option<&[ChangeHash]> {
match self {
Self::Heads(heads) => Some(heads),
Self::Digest { .. } => None,
@@ -84,7 +84,7 @@ impl HeadsSummary {
}
/// The heads listed: all of them, or a sample.
- pub fn listed(&self) -> &[ChangeHash] {
+ pub(crate) fn listed(&self) -> &[ChangeHash] {
match self {
Self::Heads(heads) => heads,
Self::Digest { sample, .. } => sample,
@@ -114,23 +114,23 @@ fn digest_of(sorted_heads: &[ChangeHash]) -> [u8; 32] {
/// PING and PONG. Addressed to one peer: `syncs_with_you` is about the recipient.
#[derive(Debug, Clone, Serialize, Deserialize)]
-pub struct Keepalive {
+pub(crate) struct Keepalive {
/// Only comparable between nodes on the same epoch.
#[serde(with = "crate::transport::wire::codec::heads")]
- pub heads: HeadsSummary,
+ pub(crate) heads: HeadsSummary,
#[serde(with = "crate::transport::wire::codec::epoch")]
- pub epoch: Epoch,
+ pub(crate) epoch: Epoch,
/// See `SyncEnvelope::epoch_seq`.
- pub epoch_seq: u32,
+ pub(crate) epoch_seq: u32,
/// The sender is ready (same meaning as `SyncEnvelope::ready`).
- pub ready: bool,
+ pub(crate) ready: bool,
/// The sender would sync with the recipient: counts it as an active member on its
/// own epoch.
- pub syncs_with_you: bool,
+ pub(crate) syncs_with_you: bool,
}
#[derive(Debug)]
-pub enum ParsedMessage {
+pub(crate) enum ParsedMessage {
Hello(HelloRequest),
HelloReply(HelloReply),
Sync(SyncEnvelope),
@@ -142,7 +142,7 @@ pub enum ParsedMessage {
/// Content-addressed bytes that any node holding them can serve, in chunks
/// (`swarm::blob`).
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)]
-pub enum BlobId {
+pub(crate) enum BlobId {
/// An Automerge change, by its hash: e.g. an epoch's snapshot change.
Change(#[serde(with = "crate::transport::wire::codec::change_hash")] ChangeHash),
/// Bytes by their BLAKE3 hash: e.g. the content of a remake.
@@ -164,18 +164,18 @@ impl std::fmt::Display for BlobId {
/// A blob request or one of its answers. Each is a message of its own, so no message
/// is larger than one chunk.
#[derive(Debug, Clone, Serialize, Deserialize)]
-pub struct BlobMessage {
+pub(crate) struct BlobMessage {
/// The sender's epoch (every message carries it).
#[serde(with = "crate::transport::wire::codec::epoch")]
- pub epoch: Epoch,
+ pub(crate) epoch: Epoch,
/// See `SyncEnvelope::epoch_seq`.
- pub epoch_seq: u32,
- pub id: BlobId,
- pub body: BlobBody,
+ pub(crate) epoch_seq: u32,
+ pub(crate) id: BlobId,
+ pub(crate) body: BlobBody,
}
#[derive(Clone, Serialize, Deserialize)]
-pub enum BlobBody {
+pub(crate) enum BlobBody {
/// Asks for these chunks; for the manifest if empty.
Request { chunks: Vec<u32> },
/// Size and per-chunk BLAKE3 hashes; the last chunk may be shorter.
@@ -223,7 +223,7 @@ impl std::fmt::Debug for BlobBody {
/// What a message is, without its contents.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
-pub enum MessageKind {
+pub(crate) enum MessageKind {
Hello,
HelloReply,
Sync,
@@ -233,7 +233,7 @@ pub enum MessageKind {
}
impl ParsedMessage {
- pub fn kind(&self) -> MessageKind {
+ pub(crate) fn kind(&self) -> MessageKind {
match self {
ParsedMessage::Hello(_) => MessageKind::Hello,
ParsedMessage::HelloReply(_) => MessageKind::HelloReply,
@@ -245,7 +245,7 @@ impl ParsedMessage {
}
/// Every message carries the sender's epoch.
- pub fn epoch(&self) -> Epoch {
+ pub(crate) fn epoch(&self) -> Epoch {
match self {
ParsedMessage::Hello(hello) => hello.epoch,
ParsedMessage::HelloReply(reply) => reply.epoch,
@@ -256,7 +256,7 @@ impl ParsedMessage {
}
/// `SyncEnvelope::epoch_seq`, as every message carries it.
- pub fn epoch_seq(&self) -> u32 {
+ pub(crate) fn epoch_seq(&self) -> u32 {
match self {
ParsedMessage::Hello(hello) => hello.epoch_seq,
ParsedMessage::HelloReply(reply) => reply.epoch_seq,
@@ -268,83 +268,83 @@ impl ParsedMessage {
}
#[derive(Debug, Clone, Serialize, Deserialize)]
-pub struct HelloRequest {
+pub(crate) struct HelloRequest {
/// The sender's swarm, if it knows it (a new node doesn't yet). A receiver in
/// another swarm doesn't answer.
- pub swarm_id: Option<String>,
+ pub(crate) swarm_id: Option<String>,
#[serde(with = "crate::transport::wire::codec::epoch")]
- pub epoch: Epoch,
+ pub(crate) epoch: Epoch,
/// See `SyncEnvelope::epoch_seq`.
- pub epoch_seq: u32,
+ pub(crate) epoch_seq: u32,
/// Proof of the swarm key (`swarm::auth`); empty without one.
#[serde(with = "serde_bytes")]
- pub auth: Vec<u8>,
+ pub(crate) auth: Vec<u8>,
/// The sender is ready (same meaning as `SyncEnvelope::ready`).
- pub ready: bool,
+ pub(crate) ready: bool,
/// Random per run of the sender; the reply echoes it.
#[serde(with = "crate::transport::wire::codec::run_id")]
- pub run_id: RunId,
+ pub(crate) run_id: RunId,
/// The handshake's number: the sender numbers the handshakes it starts in a run,
/// with all peers, from 1. A retry carries the same number, so the recipient resets
/// its sync session only for a new handshake, not for every resend or late copy.
- pub handshake: u32,
+ pub(crate) handshake: u32,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
-pub struct HelloReply {
- pub swarm_id: String,
- pub peers: Vec<String>,
+pub(crate) struct HelloReply {
+ pub(crate) swarm_id: String,
+ pub(crate) peers: Vec<String>,
#[serde(with = "crate::transport::wire::codec::epoch")]
- pub epoch: Epoch,
+ pub(crate) epoch: Epoch,
/// See `SyncEnvelope::epoch_seq`.
- pub epoch_seq: u32,
+ pub(crate) epoch_seq: u32,
/// Proof of the swarm key (`swarm::auth`); empty without one.
#[serde(with = "serde_bytes")]
- pub auth: Vec<u8>,
+ pub(crate) auth: Vec<u8>,
/// The sender is ready.
- pub ready: bool,
+ pub(crate) ready: bool,
/// The `run_id` of the HELLO answered. A reply to a HELLO from an earlier run of the
/// recipient (delivered late, or queued while it was down) is ignored.
#[serde(with = "crate::transport::wire::codec::run_id")]
- pub run_id: RunId,
+ pub(crate) run_id: RunId,
/// How long before this HELLO the sender had last heard from the recipient, if ever
/// (in milliseconds): tells a restarted node how long it was away.
- pub away_ms: Option<u64>,
+ pub(crate) away_ms: Option<u64>,
/// The sender's own run (as its HELLOs carry it): a new one means it restarted.
#[serde(with = "crate::transport::wire::codec::run_id")]
- pub sender_run_id: RunId,
+ pub(crate) sender_run_id: RunId,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
-pub struct SyncEnvelope {
- pub swarm_id: String,
+pub(crate) struct SyncEnvelope {
+ pub(crate) swarm_id: String,
/// Epoch of the document the message was generated from.
#[serde(with = "crate::transport::wire::codec::epoch")]
- pub epoch: Epoch,
+ pub(crate) epoch: Epoch,
/// How many times the sender changed epoch in its current run, when it sent this
/// (`Replica::epoch_seq`). Messages may arrive late and out of order: the receiver
/// ignores an epoch older than the newest it has seen in the same session.
- pub epoch_seq: u32,
+ pub(crate) epoch_seq: u32,
#[serde(with = "crate::transport::wire::codec::sync_message")]
- pub message: SyncPayload,
+ pub(crate) message: SyncPayload,
/// The sender was ready when it generated the message: its document is the swarm's,
/// not one it is still filling (joining, rejoining). Only such a peer can tell a
/// catching-up node that it has everything; an empty document says nothing else.
- pub ready: bool,
+ pub(crate) ready: bool,
}
/// An Automerge sync message in Automerge's own encoding. Decoded only where it is
/// applied (`decode`), after the sender proved to be a member and off the consumer: a
/// large message from anyone else costs nothing but its bytes.
#[derive(Clone)]
-pub struct SyncPayload(pub bytes::Bytes);
+pub(crate) struct SyncPayload(pub(crate) bytes::Bytes);
impl SyncPayload {
- pub fn encode(message: Message) -> Self {
+ pub(crate) fn encode(message: Message) -> Self {
Self(message.encode().into())
}
- pub fn decode(&self) -> Result<Message> {
+ pub(crate) fn decode(&self) -> Result<Message> {
Message::decode(&self.0)
.map_err(|error| anyhow::anyhow!("Automerge decode failed: {error:?}"))
}
@@ -359,23 +359,23 @@ impl std::fmt::Debug for SyncPayload {
/// 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 kind: MessageKind,
+pub(crate) struct EncodedMessage {
+ pub(crate) peer: String,
+ pub(crate) body: bytes::Bytes,
+ pub(crate) kind: MessageKind,
/// Bytes the message takes on the wire (without the transport's own framing).
- pub payload_len: usize,
+ pub(crate) payload_len: usize,
}
/// Called with the sender, the message and its `payload_len` (as in `EncodedMessage`).
-pub type HandlerFn = dyn Fn(String, ParsedMessage, usize) -> Pin<Box<dyn Future<Output = Result<()>> + Send>>
+pub(crate) type HandlerFn = dyn Fn(String, ParsedMessage, usize) -> Pin<Box<dyn Future<Output = Result<()>> + Send>>
+ Send
+ Sync
+ 'static;
-pub type MessageHandler = Box<HandlerFn>;
+pub(crate) type MessageHandler = Box<HandlerFn>;
#[async_trait]
-pub trait Transport: Send + Sync {
+pub(crate) trait Transport: Send + Sync {
/// This node's identity on the transport (for Mycelium: its public key).
fn local_id(&self) -> &str;
diff --git a/src/transport/mycelium/daemon.rs b/src/transport/mycelium/daemon.rs
index f194fca..f986d52 100644
--- a/src/transport/mycelium/daemon.rs
+++ b/src/transport/mycelium/daemon.rs
@@ -14,28 +14,28 @@ use url::Url;
/// Represents where a 32-byte cryptographic key originates.
#[derive(Debug, Clone)]
-pub enum KeySource {
+pub(crate) enum KeySource {
Raw([u8; 32]),
Path(PathBuf),
}
/// Atomic structure ensuring private network credentials are always handled together.
#[derive(Debug, Clone)]
-pub struct PrivateNetworkConfig {
- pub name: String,
- pub key: KeySource,
+pub(crate) struct PrivateNetworkConfig {
+ pub(crate) name: String,
+ pub(crate) key: KeySource,
}
/// Comprehensive parameters required to spawn or target an instance.
#[derive(Debug, Clone)]
-pub struct MyceliumSpawnConfig {
- pub rest_url_opt: Option<Url>,
- pub key_file_opt: Option<PathBuf>,
- pub network_key_file: Option<PathBuf>,
+pub(crate) struct MyceliumSpawnConfig {
+ pub(crate) rest_url_opt: Option<Url>,
+ pub(crate) key_file_opt: Option<PathBuf>,
+ pub(crate) network_key_file: Option<PathBuf>,
/// Set together with `network_key_file` (`Settings::check`).
- pub network_name: Option<String>,
- pub tcp_port: Option<u16>,
- pub peers: Vec<String>,
+ pub(crate) network_name: Option<String>,
+ pub(crate) tcp_port: Option<u16>,
+ pub(crate) peers: Vec<String>,
}
impl Default for MyceliumSpawnConfig {
@@ -51,19 +51,19 @@ impl Default for MyceliumSpawnConfig {
}
}
-pub enum MyceliumRuntime {
+pub(crate) enum MyceliumRuntime {
Supervised(Box<SupervisedMycelium>),
External,
}
-pub struct MyceliumSetup {
- pub runtime: MyceliumRuntime,
- pub rest_url: Url,
+pub(crate) struct MyceliumSetup {
+ pub(crate) runtime: MyceliumRuntime,
+ pub(crate) rest_url: Url,
}
impl MyceliumSetup {
/// Evaluates the configuration payload and initializes the chosen execution strategy.
- pub async fn bootstrap(config: MyceliumSpawnConfig) -> Result<Self> {
+ pub(crate) async fn bootstrap(config: MyceliumSpawnConfig) -> Result<Self> {
// TODO warn for all ignored options if rest_url_opt is set
if let Some(url) = config.rest_url_opt {
tracing::info!(
@@ -115,7 +115,7 @@ impl MyceliumSetup {
}
/// Blocks until the background daemon drops out. If running externally, hangs open indefinitely.
- pub async fn wait_for_exit(&mut self) -> std::io::Result<std::process::ExitStatus> {
+ pub(crate) async fn wait_for_exit(&mut self) -> std::io::Result<std::process::ExitStatus> {
match &mut self.runtime {
MyceliumRuntime::Supervised(node) => node.wait_for_exit().await,
MyceliumRuntime::External => {
@@ -126,16 +126,16 @@ impl MyceliumSetup {
}
#[derive(Debug, Clone)]
-pub struct MyceliumNodeConfig {
- pub tcp_port: u16,
- pub api_port: u16,
- pub rpc_port: u16,
- pub private_network: Option<String>,
+pub(crate) struct MyceliumNodeConfig {
+ pub(crate) tcp_port: u16,
+ pub(crate) api_port: u16,
+ pub(crate) rpc_port: u16,
+ pub(crate) private_network: Option<String>,
}
-pub struct SupervisedMycelium {
+pub(crate) struct SupervisedMycelium {
child: Child,
- pub config: MyceliumNodeConfig,
+ pub(crate) config: MyceliumNodeConfig,
_key_file_handle: Option<File>,
_net_key_file_handle: Option<File>,
}
@@ -327,7 +327,7 @@ impl SupervisedMycelium {
}
}
- pub async fn wait_for_exit(&mut self) -> std::io::Result<std::process::ExitStatus> {
+ pub(crate) async fn wait_for_exit(&mut self) -> std::io::Result<std::process::ExitStatus> {
self.child.wait().await
}
}
diff --git a/src/transport/mycelium/mod.rs b/src/transport/mycelium/mod.rs
index 5062d4a..6304819 100644
--- a/src/transport/mycelium/mod.rs
+++ b/src/transport/mycelium/mod.rs
@@ -10,9 +10,9 @@ use url::Url;
use crate::transport::wire;
use crate::transport::{EncodedMessage, MessageHandler, ParsedMessage, Transport};
-pub mod daemon;
+pub(crate) mod daemon;
-pub use daemon::{MyceliumSetup, MyceliumSpawnConfig};
+pub(crate) use daemon::{MyceliumSetup, MyceliumSpawnConfig};
const MYCELIUM_POLL_TIMEOUT_SECS: u64 = 300;
/// The largest inbound message read (its JSON, base64 payload included: about 190 MiB of
@@ -64,15 +64,15 @@ struct Destination {
/// The part of Mycelium's `/api/v1/admin` answer we use.
#[derive(Deserialize, Debug, Clone)]
#[serde(rename_all = "camelCase")]
-pub struct MyceliumAdminInfo {
- pub node_pubkey: String,
+pub(crate) struct MyceliumAdminInfo {
+ pub(crate) node_pubkey: String,
}
fn is_pubkey(s: &str) -> bool {
s.len() == 64 && s.chars().all(|c| c.is_ascii_hexdigit())
}
-pub struct MyceliumTransport {
+pub(crate) struct MyceliumTransport {
/// The daemon, kept to wait for its exit.
setup: tokio::sync::Mutex<MyceliumSetup>,
client: reqwest::Client,
@@ -84,7 +84,7 @@ pub struct MyceliumTransport {
}
impl MyceliumTransport {
- pub async fn connect(setup: MyceliumSetup) -> Result<Self> {
+ pub(crate) async fn connect(setup: MyceliumSetup) -> Result<Self> {
let rest_url = setup.rest_url.clone();
let client = reqwest::Client::builder()
@@ -515,7 +515,7 @@ fn format_message(
}
#[derive(Error, Debug)]
-pub enum TransportError {
+pub(crate) enum TransportError {
#[error("Mycelium delivered a message without a sender public key")]
NoSender,
diff --git a/src/transport/traffic.rs b/src/transport/traffic.rs
index 7d245ac..6c3d683 100644
--- a/src/transport/traffic.rs
+++ b/src/transport/traffic.rs
@@ -19,9 +19,9 @@ use crate::locks::MutexExt;
use crate::transport::{EncodedMessage, MessageHandler, MessageKind, ParsedMessage, Transport};
#[derive(Debug, Default, Clone, Copy, Serialize)]
-pub struct Counter {
- pub messages: u64,
- pub bytes: u64,
+pub(crate) struct Counter {
+ pub(crate) messages: u64,
+ pub(crate) bytes: u64,
}
impl Counter {
@@ -33,13 +33,13 @@ impl Counter {
#[derive(Debug, Default, Clone, Copy, Serialize)]
#[serde(rename_all = "camelCase")]
-pub struct ByKind {
- pub hello: Counter,
- pub hello_reply: Counter,
- pub sync: Counter,
- pub ping: Counter,
- pub pong: Counter,
- pub blob: Counter,
+pub(crate) struct ByKind {
+ pub(crate) hello: Counter,
+ pub(crate) hello_reply: Counter,
+ pub(crate) sync: Counter,
+ pub(crate) ping: Counter,
+ pub(crate) pong: Counter,
+ pub(crate) blob: Counter,
}
impl ByKind {
@@ -57,12 +57,12 @@ impl ByKind {
#[derive(Debug, Default, Clone, Copy, Serialize)]
#[serde(rename_all = "camelCase")]
-pub struct PeerTraffic {
+pub(crate) struct PeerTraffic {
/// Handed to Mycelium (it accepted them).
- pub sent: ByKind,
- pub received: ByKind,
+ pub(crate) sent: ByKind,
+ pub(crate) received: ByKind,
/// Sends Mycelium refused (e.g. no route to the peer); not counted as sent.
- pub refused: ByKind,
+ pub(crate) refused: ByKind,
}
#[derive(Debug, Default)]
@@ -73,7 +73,7 @@ struct Table {
}
#[derive(Debug, Default)]
-pub struct Traffic {
+pub(crate) struct Traffic {
table: Mutex<Table>,
}
@@ -104,29 +104,29 @@ impl Traffic {
self.count(peer, bytes, |t| t.received.get_mut(kind));
}
- pub fn peer(&self, peer: &str) -> Option<PeerTraffic> {
+ pub(crate) fn peer(&self, peer: &str) -> Option<PeerTraffic> {
self.lock().peers.get(peer).copied()
}
- pub fn total(&self) -> PeerTraffic {
+ pub(crate) fn total(&self) -> PeerTraffic {
self.lock().total
}
/// Drops the counters of peers not kept (anyone can send us messages; the totals
/// keep their share).
- pub fn retain(&self, mut keep: impl FnMut(&str) -> bool) {
+ pub(crate) fn retain(&self, mut keep: impl FnMut(&str) -> bool) {
self.lock().peers.retain(|peer, _| keep(peer));
}
}
/// Counts what passes through `inner`.
-pub struct CountingTransport {
+pub(crate) struct CountingTransport {
inner: Arc<dyn Transport>,
traffic: Arc<Traffic>,
}
impl CountingTransport {
- pub fn new(inner: Arc<dyn Transport>, traffic: Arc<Traffic>) -> Self {
+ pub(crate) fn new(inner: Arc<dyn Transport>, traffic: Arc<Traffic>) -> Self {
Self { inner, traffic }
}
}
diff --git a/src/transport/wire.rs b/src/transport/wire.rs
index a0e0f20..67d2fb3 100644
--- a/src/transport/wire.rs
+++ b/src/transport/wire.rs
@@ -29,7 +29,7 @@ fn tag(kind: MessageKind) -> u8 {
}
#[derive(Error, Debug)]
-pub enum WireError {
+pub(crate) enum WireError {
#[error("empty payload")]
Empty,
generated by cgit v1.3.1 (git 2.54.0) at 2026-10-28 06:25:06 +0000