use std::{cell::RefCell, collections::HashMap, pin::Pin, rc::Rc};
use audi::{Listener, ListenerSet};
use endr::ObjectID;
use futures::{stream, Future, FutureExt, Stream, StreamExt};
use jmbl::{ops::OpWithTarget, JMBLViewRef, OpOutput, JMBL, SrcID, Value};
use tracing::trace;
use super::{ContentDiff, ContentType};
use crate::{doc::WeakDoc, Doc};
pub struct JMBLContentInner {
weak_doc: Option<WeakDoc>,
jmbl: JMBL,
change_callbacks: Vec<Box<dyn FnMut(jmbl::Value, bool)>>,
listeners: ListenerSet<(jmbl::Value, bool)>,
jmbl_srcs_to_endr_logs: HashMap<jmbl::SrcID, endr::ObjectID>,
last_write_log: Option<endr::ObjectID>,
creation_input: Option<Box<dyn FnOnce(JMBLViewRef) -> Value>>,
}
impl JMBLContentInner {
fn receive_ops(&mut self, ops: &[OpWithTarget], from_endr_log: ObjectID) {
self.jmbl.apply_ops(ops);
if let Some(op) = ops.first() {
self.jmbl_srcs_to_endr_logs
.entry(op.op.id.0)
.or_insert(from_endr_log);
}
for callback in &mut self.change_callbacks {
callback(self.jmbl.get_root(), true);
}
}
}
#[derive(Clone)]
pub struct JMBLContent(Rc<RefCell<JMBLContentInner>>);
impl ContentType for JMBLContent {
fn content_type(&self) -> &'static str {
"jmbl1"
}
fn connect_and_init(&self, weak_doc: WeakDoc, require_intro: bool) {
self.0.borrow_mut().weak_doc = Some(weak_doc);
let maybe_input = self.0.borrow_mut().creation_input.take();
if let Some(constructor) = maybe_input {
let mut write_view = self.start_writing(require_intro);
let root = constructor(write_view.clone());
write_view.set_root(root.obj_id().unwrap());
self.finish_writing(write_view);
}
}
fn follow_new_log(
&self,
log_id: ObjectID,
diffs: Pin<Box<dyn Stream<Item = ContentDiff>>>,
) -> Pin<Box<dyn Future<Output = ()>>> {
let self_bg = self.clone();
reading::diffs_to_ops(diffs)
.ready_chunks(10000)
.for_each(move |ops| {
trace!(n_ops = ops.len(), "Received ops");
(*self_bg.0).borrow_mut().receive_ops(&ops, log_id);
let root = self_bg.current_root();
let listeners = (*self_bg.0).borrow().listeners.clone();
async move {
listeners.broadcast((root, true)).await;
}
})
.boxed_local()
}
}
impl JMBLContent {
pub fn new_empty() -> Self {
JMBLContent(Rc::new(RefCell::new(JMBLContentInner {
weak_doc: None,
jmbl: JMBL::new_empty(),
change_callbacks: Vec::new(),
listeners: ListenerSet::new(),
jmbl_srcs_to_endr_logs: HashMap::new(),
last_write_log: None,
creation_input: None,
})))
}
pub fn create<C: FnOnce(JMBLViewRef) -> Value + 'static>(constructor: C) -> Self {
let content = JMBLContent::new_empty();
content.0.borrow_mut().creation_input = Some(Box::new(constructor));
content
}
pub fn doc(&self) -> Option<Doc> {
self.0.borrow().weak_doc.as_ref().and_then(|weak_doc| weak_doc.upgrade())
}
pub fn current_root(&self) -> jmbl::Value {
(*self.0).borrow().jmbl.get_root()
}
pub fn add_change_callback<F>(&self, mut callback: F)
where
F: FnMut(jmbl::Value, bool) + 'static,
{
callback(self.current_root(), false);
(*self.0)
.borrow_mut()
.change_callbacks
.push(Box::new(callback));
}
async fn add_listener(&self, listener: Listener<(jmbl::Value, bool)>) {
let listeners = (*self.0).borrow().listeners.clone();
listeners
.add_with_initial_msg(listener, Some((self.current_root(), false)))
.await;
}
pub fn updates(
&self,
listener_prefix: String,
) -> impl Stream<Item = (jmbl::Value, bool)> + 'static {
let self_rc = self.clone();
stream::once(async move {
let (tx, rx) = futures::channel::mpsc::channel(100);
let listener = Listener::new(
&format!("{}_{:?}", listener_prefix, rand07::random::<u64>()),
tx,
);
self_rc.add_listener(listener).await;
rx
})
.flatten()
.boxed_local()
}
pub fn start_writing(&self, require_into: bool) -> JMBLViewRef {
let doc: Doc = (*self.0)
.borrow()
.weak_doc
.as_ref()
.expect("Expected doc content to be connected to doc when starting to write")
.upgrade()
.unwrap();
let (new_data_tx, new_log_id) = if require_into {doc.start_writing()} else {doc.start_writing_without_intro()};
if (*self.0).borrow().last_write_log != Some(new_log_id) {
let mut self_ref = (*self.0).borrow_mut();
let new_output = OpOutput::new(Box::new(writing::DocOpSender::new(
new_data_tx
)));
self_ref
.jmbl_srcs_to_endr_logs
.insert(new_output.src_id, new_log_id);
self_ref.jmbl = self_ref.jmbl.switch_to_new_op_output(new_output);
self_ref.last_write_log = Some(new_log_id);
}
return (*self.0).borrow_mut().jmbl.start_changing_object();
}
pub fn finish_writing(&self, view: JMBLViewRef) {
(*self.0).borrow_mut().jmbl.finish_changing_object(view);
}
pub fn get_endr_log_for_jmbl_log(&self, jmbl_log: SrcID) -> Option<ObjectID> {
(*self.0)
.borrow()
.jmbl_srcs_to_endr_logs
.get(&jmbl_log)
.cloned()
}
}
mod reading;
mod writing;