syndicate/
dataspace.rs

1//! Implements a [*dataspace*][crate::dataspace#GarnockJones2017]
2//! entity.
3//!
4//! **References.**
5//!
6//! - Garnock-Jones, Tony. <a name="GarnockJones2017">“Conversational
7//!   Concurrency.”</a> PhD, Northeastern University, 2017. [Available
8//!   on the web](https://syndicate-lang.org/tonyg-dissertation/).
9//!   [PDF](https://syndicate-lang.org/papers/conversational-concurrency-201712310922.pdf).
10
11use super::language;
12use super::skeleton;
13use super::actor::*;
14use super::schemas::dataspace::*;
15
16use preserves::value::Map;
17use preserves_schema::Codec;
18
19/// A Dataspace object (entity).
20#[derive(Debug)]
21pub struct Dataspace {
22    pub name: Name,
23    /// Index over assertions placed in the dataspace; used to
24    /// efficiently route assertion changes and messages to observers.
25    pub index: skeleton::Index,
26    /// Local memory of assertions indexed by `Handle`, used to remove
27    /// assertions from the `index` when they are retracted.
28    pub handle_map: Map<Handle, _Any>,
29}
30
31impl Dataspace {
32    /// Construct a new, empty dataspace.
33    pub fn new(name: Name) -> Self {
34        Self {
35            name,
36            index: skeleton::Index::new(),
37            handle_map: Map::new(),
38        }
39    }
40
41    /// Retrieve the current count of *distinct* assertions placed in
42    /// the dataspace.
43    pub fn assertion_count(&self) -> usize {
44        self.index.assertion_count()
45    }
46
47    /// Retrieve the current count of assertions, including
48    /// duplicates, placed in the dataspace.
49    pub fn endpoint_count(&self) -> isize {
50        self.index.endpoint_count()
51    }
52
53    /// Retrieve the current count of [`Observe`] assertions in the dataspace.
54    pub fn observer_count(&self) -> usize {
55        self.index.observer_count()
56    }
57}
58
59impl Entity<_Any> for Dataspace {
60    fn assert(&mut self, t: &mut Activation, a: _Any, h: Handle) -> ActorResult {
61        let is_new = self.index.insert(t, &a);
62        tracing::trace!(dataspace = ?self.name, assertion = ?a, handle = ?h, ?is_new, "assert");
63
64        if is_new {
65            if let Ok(o) = language().parse::<Observe>(&a) {
66                self.index.add_observer(t, &o.pattern, &o.observer);
67            }
68        }
69
70        self.handle_map.insert(h, a);
71        Ok(())
72    }
73
74    fn retract(&mut self, t: &mut Activation, h: Handle) -> ActorResult {
75        match self.handle_map.remove(&h) {
76            None => tracing::warn!(dataspace = ?self.name, handle = ?h, "retract of unknown handle"),
77            Some(a) => {
78                let is_last = self.index.remove(t, &a);
79                tracing::trace!(dataspace = ?self.name, assertion = ?a, handle = ?h, ?is_last, "retract");
80
81                if is_last {
82                    if let Ok(o) = language().parse::<Observe>(&a) {
83                        self.index.remove_observer(t, o.pattern, &o.observer);
84                    }
85                }
86            }
87        }
88        Ok(())
89    }
90
91    fn message(&mut self, t: &mut Activation, m: _Any) -> ActorResult {
92        tracing::trace!(dataspace = ?self.name, body = ?m, "message");
93        self.index.send(t, &m);
94        Ok(())
95    }
96}
syndicate/dataspace.rs

syndicate/
dataspace.rs
