tlpt 0.8.0

A set of protocols for building local-first distributed systems
Documentation 
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
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119

use std::{cell::RefCell, collections::HashMap, rc::Rc};

use audi::{Listener, ListenerSet};
use endr::ObjectID;
use futures::{channel::mpsc::channel, future, stream, FutureExt, SinkExt, StreamExt};

use crate::ContentType;

struct JsonStreamContentInner {
    weak_doc: Option<crate::doc::WeakDoc>,
    latest_per_log: HashMap<ObjectID, litl::Val>,
    change_listeners: ListenerSet<HashMap<ObjectID, litl::Val>>,
}

#[derive(Clone)]
pub struct JsonStreamContent(Rc<RefCell<JsonStreamContentInner>>);

impl JsonStreamContent {
    pub fn new_empty() -> JsonStreamContent {
        JsonStreamContent(Rc::new(RefCell::new(JsonStreamContentInner {
            weak_doc: None,
            latest_per_log: HashMap::new(),
            change_listeners: ListenerSet::new(),
        })))
    }

    pub fn get_latest(&self, log_id: &ObjectID) -> Option<litl::Val> {
        self.0.borrow().latest_per_log.get(log_id).cloned()
    }

    pub async fn add_update_listener(&self, listener: Listener<HashMap<ObjectID, litl::Val>>) {
        let listeners = self.0.borrow().change_listeners.clone();
        let latest_per_log = self.0.borrow().latest_per_log.clone();

        listeners
            .add_with_initial_msg(listener, Some(latest_per_log))
            .await;
    }

    pub fn updates(
        &self,
        listener_prefix: String,
    ) -> impl futures::Stream<Item = HashMap<ObjectID, litl::Val>> {
        let (updates_tx, updates_rx) = channel(1000000);

        let self_clone = self.clone();

        stream::once(async move {
            self_clone
                .add_update_listener(Listener::new(
                    &format!("{}_{:?}", listener_prefix, rand07::random::<u64>()),
                    updates_tx,
                ))
                .await;

            updates_rx
        })
        .flatten()
        .boxed_local()
    }

    pub async fn add_val(&self, val: litl::Val) {
        // TODO: add compression
        let doc = self
            .0
            .borrow()
            .weak_doc
            .as_ref()
            .unwrap()
            .upgrade()
            .unwrap();

        let (mut out, _) = doc.start_writing();

        out.send(val).await.unwrap();
    }
}

impl ContentType for JsonStreamContent {
    fn content_type(&self) -> &'static str {
        return "json_stream1";
    }

    fn connect_and_init(&self, weak_doc: crate::doc::WeakDoc, _require_intro: bool) {
        let mut self_ref = self.0.borrow_mut();

        self_ref.weak_doc = Some(weak_doc);
    }

    fn follow_new_log(
        &self,
        log_id: endr::ObjectID,
        diffs: std::pin::Pin<Box<dyn futures::Stream<Item = super::ContentDiff>>>,
    ) -> std::pin::Pin<Box<dyn futures::Future<Output = ()>>> {
        let self_for_stream = self.clone();

        diffs
            .filter(|diff| future::ready(diff.decrypted_entries.len() > 0))
            .for_each(move |diff| {
                // TODO: add compression
                let self_for_stream = self_for_stream.clone();
                async move {
                    if let Some(last_val) = diff.decrypted_entries.last() {
                        self_for_stream
                            .0
                            .borrow_mut()
                            .latest_per_log
                            .insert(log_id.clone(), last_val.clone());

                        let listeners = self_for_stream.0.borrow().change_listeners.clone();
                        let latest_per_log = self_for_stream.0.borrow().latest_per_log.clone();

                    listeners.broadcast(latest_per_log).await;
                    }
                }
            })
            .boxed_local()
    }
}