use std::collections::{HashMap, HashSet};
use std::sync::Arc;
use automerge::{Automerge, ReadDoc};
use futures::{Stream, channel::mpsc};
use samod_core::{
ConnectionId, DocumentActorId, DocumentChanged, DocumentId, PeerId, SyncDirection,
SyncMessageStat, UnixTimestamp,
actors::{
DocToHubMsg,
document::{DocActorResult, DocumentActor, WithDocResult},
},
};
use crate::{
actor_task::ActorTask,
io_loop::{self, IoLoopTask},
observer::{self, RepoObserver},
peer_connection_info::PeerDocState,
unbounded::UnboundedSender,
};
pub(crate) struct DocActorInner {
document_id: DocumentId,
actor_id: DocumentActorId,
tx_to_core: UnboundedSender<(DocumentActorId, DocToHubMsg)>,
tx_io: UnboundedSender<io_loop::IoLoopTask>,
ephemera_listeners: Vec<mpsc::UnboundedSender<Vec<u8>>>,
change_listeners: Vec<mpsc::UnboundedSender<DocumentChanged>>,
peer_state_change_listeners: Vec<mpsc::UnboundedSender<HashMap<ConnectionId, PeerDocState>>>,
actor: DocumentActor,
last_observed_peers: HashSet<PeerId>,
observer: Option<Arc<dyn RepoObserver>>,
}
impl DocActorInner {
pub(crate) fn new(
document_id: DocumentId,
actor_id: DocumentActorId,
actor: DocumentActor,
tx_to_core: UnboundedSender<(DocumentActorId, DocToHubMsg)>,
tx_io: UnboundedSender<io_loop::IoLoopTask>,
observer: Option<Arc<dyn RepoObserver>>,
) -> Self {
DocActorInner {
document_id,
actor_id,
tx_to_core,
tx_io,
ephemera_listeners: Vec::new(),
change_listeners: Vec::new(),
peer_state_change_listeners: Vec::new(),
actor,
last_observed_peers: HashSet::new(),
observer,
}
}
pub(crate) fn document(&self) -> &Automerge {
self.actor.document()
}
pub(crate) fn with_document<F, R>(&mut self, f: F) -> R
where
F: FnOnce(&mut Automerge) -> R,
{
let WithDocResult {
actor_result,
value,
} = self.actor.with_document(UnixTimestamp::now(), f).unwrap();
self.handle_results(actor_result);
value
}
pub(crate) fn create_ephemera_listener(&mut self) -> mpsc::UnboundedReceiver<Vec<u8>> {
let (tx, rx) = mpsc::unbounded();
self.ephemera_listeners.push(tx);
rx
}
pub(crate) fn create_change_listener(&mut self) -> mpsc::UnboundedReceiver<DocumentChanged> {
let (tx, rx) = mpsc::unbounded();
self.change_listeners.push(tx);
rx
}
pub(crate) fn broadcast_ephemeral_message(&mut self, message: Vec<u8>) {
let result = self.actor.broadcast(UnixTimestamp::now(), message);
self.handle_results(result);
}
pub(crate) fn handle_results(&mut self, results: DocActorResult) {
let DocActorResult {
io_tasks,
outgoing_messages,
ephemeral_messages,
change_events,
stopped,
peer_state_changes,
sync_message_stats,
pending_sync_messages,
} = results;
if let Some(new_peers) = self.connected_peers_changed(stopped) {
let _ = self.tx_io.unbounded_send(IoLoopTask::DocumentPeersChanged {
actor_id: self.actor_id,
peers: new_peers,
});
}
let peer_count = self.actor.peers().len();
let stats = self.actor.document().stats();
log_sync_message_stats(
&self.document_id,
&sync_message_stats,
peer_count,
stats.num_ops,
stats.num_changes,
);
if let Some(ref obs) = self.observer {
for stat in &sync_message_stats {
let event = match stat.direction {
SyncDirection::Received => observer::RepoEvent::SyncMessageReceived {
document_id: self.document_id.clone(),
connection_id: stat.connection_id,
bytes: stat.bytes,
duration: stat.duration,
queue_duration: stat.queue_duration,
},
SyncDirection::Generated => observer::RepoEvent::SyncMessageGenerated {
document_id: self.document_id.clone(),
connection_id: stat.connection_id,
bytes: stat.bytes,
duration: stat.duration,
},
};
obs.observe(&event);
}
if pending_sync_messages > 0 {
obs.observe(&observer::RepoEvent::DocumentPendingSyncMessages {
document_id: self.document_id.clone(),
count: pending_sync_messages,
});
}
if stopped {
obs.observe(&observer::RepoEvent::DocumentClosed {
document_id: self.document_id.clone(),
});
}
}
for task in io_tasks {
if let Err(_e) = self.tx_io.unbounded_send(IoLoopTask::Storage {
doc_id: self.document_id.clone(),
task,
actor_id: self.actor_id,
}) {
tracing::error!("io receiver dropped whilst document actor is still running");
return;
}
}
for msg in outgoing_messages {
if let Err(_e) = self.tx_to_core.unbounded_send((self.actor_id, msg)) {
tracing::error!("core receiver dropped whilst document actor is still running");
return;
}
}
if !ephemeral_messages.is_empty() {
self.ephemera_listeners.retain_mut(|listener| {
for msg in &ephemeral_messages {
if listener.unbounded_send(msg.clone()).is_err() {
return false;
}
}
true
});
}
if !change_events.is_empty() {
self.change_listeners.retain_mut(|listener| {
for change in &change_events {
if listener.unbounded_send(change.clone()).is_err() {
return false;
}
}
true
});
}
if !peer_state_changes.is_empty() {
let changes = peer_state_changes
.into_iter()
.map(|(k, v)| (k, v.into()))
.collect::<HashMap<_, _>>();
self.peer_state_change_listeners
.retain_mut(move |listener| {
if listener.unbounded_send(changes.clone()).is_err() {
return false;
}
true
});
}
}
pub(crate) fn handle_task(&mut self, task: ActorTask) {
let result = match task {
ActorTask::HandleMessage(samod_to_actor_message) => self
.actor
.handle_message(UnixTimestamp::now(), samod_to_actor_message),
ActorTask::IoComplete(io_result) => self
.actor
.handle_io_complete(UnixTimestamp::now(), io_result),
ActorTask::WithDocument(f) => {
f(self);
return;
}
};
self.handle_results(result.unwrap());
}
pub(crate) fn is_stopped(&self) -> bool {
self.actor.is_stopped()
}
pub(crate) fn peers(
&mut self,
) -> (
HashMap<ConnectionId, PeerDocState>,
impl Stream<Item = HashMap<ConnectionId, PeerDocState>> + 'static + use<>,
) {
let peers = self.actor.peers();
let (tx, rx) = mpsc::unbounded();
self.peer_state_change_listeners.push(tx);
let peers = peers.into_iter().map(|(k, v)| (k, v.into())).collect();
(peers, rx)
}
pub(crate) fn connected_peers_changed(&mut self, stopped: bool) -> Option<HashSet<PeerId>> {
let peers = self.actor.peers();
let peers = if stopped {
HashSet::new()
} else {
peers
.keys()
.filter_map(|id| self.actor.conn_peer_id(*id))
.collect()
};
if peers != self.last_observed_peers {
self.last_observed_peers = peers.clone();
Some(peers)
} else {
None
}
}
}
fn log_sync_message_stats(
doc_id: &DocumentId,
sync_message_stats: &[SyncMessageStat],
peer_count: usize,
num_ops: u64,
num_changes: u64,
) {
for stat in sync_message_stats {
let direction = match stat.direction {
SyncDirection::Received => "received",
SyncDirection::Generated => "generated",
};
tracing::debug!(
document_id = %doc_id,
connection_id = ?stat.connection_id,
direction,
bytes = stat.bytes,
duration_us = stat.duration.as_micros() as u64,
queue_duration_us = stat.queue_duration.as_micros() as u64,
peer_count,
num_ops,
num_changes,
"sync_message"
);
}
}