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
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158

use super::{managed_state::ManagedState, topic::Topic};
use crate::{
    debug::DebugTypeName,
    nut::{exec::Deferred, Handler, Nut, IMPOSSIBLE_ERR_MSG},
    ActivityId, UncheckedActivityId,
};
use core::cell::Ref;
use std::{
    any::Any,
    cell::RefCell,
    collections::HashMap,
    ops::{Index, IndexMut},
};

#[derive(Default)]
pub(crate) struct Subscriptions {
    subscriptions: RefCell<HashMap<Topic, SubscriptionContainer>>,
}

/// Handlers stored per Activity
#[derive(Default)]
pub(crate) struct SubscriptionContainer {
    data: HashMap<usize, ActivityTopicSubscriptions>,
}

/// Handlers per type per activity
#[derive(Default)]
pub(crate) struct ActivityTopicSubscriptions {
    shared: Vec<Subscription>,
    private: Option<Subscription>,
}

pub(crate) struct Subscription {
    pub(crate) handler: Handler,
    pub(crate) type_name: DebugTypeName,
}

pub(crate) enum OnDelete {
    None,
    Simple(Box<dyn FnOnce(Box<dyn Any>)>),
    WithDomain(Box<dyn FnOnce(Box<dyn Any>, &mut ManagedState)>),
}

impl Nut {
    pub(crate) fn push_closure<A: 'static>(
        &self,
        topic: Topic,
        id: ActivityId<A>,
        closure: Handler,
    ) {
        let type_name = DebugTypeName::new::<A>();
        if self.quiescent() {
            self.subscriptions
                .force_push_closure(topic, id, closure, type_name);
        } else {
            let sub = NewSubscription::new(topic, id, closure, type_name);
            self.deferred_events.push(Deferred::Subscription(sub));
        }
    }
}

impl Subscriptions {
    pub(crate) fn exec_new_subscription(&self, sub: NewSubscription) {
        self.force_push_closure(sub.topic, sub.id, sub.closure, sub.type_name);
    }
    fn force_push_closure(
        &self,
        topic: Topic,
        id: impl Into<UncheckedActivityId>,
        handler: Handler,
        type_name: DebugTypeName,
    ) {
        let id = id.into();
        let private = topic.unqiue_per_activity();
        let subs = &mut self
            .subscriptions
            .try_borrow_mut()
            .expect(IMPOSSIBLE_ERR_MSG);
        let subs_per_activity = &mut subs.entry(topic).or_insert_with(Default::default)[id];

        if private {
            subs_per_activity.private = Some(Subscription { handler, type_name });
        } else {
            subs_per_activity
                .shared
                .push(Subscription { handler, type_name });
        }
    }
    pub(crate) fn get(&self) -> Ref<HashMap<Topic, SubscriptionContainer>> {
        self.subscriptions.borrow()
    }
}

impl SubscriptionContainer {
    pub fn shared_subscriptions(&self) -> impl Iterator<Item = &Subscription> {
        self.data.values().flat_map(|f| f.shared.iter())
    }
    pub fn shared_subscriptions_of_single_activity(
        &self,
        id: UncheckedActivityId,
    ) -> impl Iterator<Item = &Subscription> {
        self.data
            .get(&id.index)
            .into_iter()
            .flat_map(|f| f.shared.iter())
    }
    pub fn private_subscription(&self, id: UncheckedActivityId) -> Option<&Subscription> {
        self.data
            .get(&id.index)
            .map(|f| f.private.as_ref())
            .flatten()
    }
}
impl Index<UncheckedActivityId> for SubscriptionContainer {
    type Output = ActivityTopicSubscriptions;
    fn index(&self, id: UncheckedActivityId) -> &Self::Output {
        &self.data[&id.index]
    }
}
impl IndexMut<UncheckedActivityId> for SubscriptionContainer {
    fn index_mut(&mut self, id: UncheckedActivityId) -> &mut Self::Output {
        self.data.entry(id.index).or_insert_with(Default::default)
    }
}

pub(crate) struct NewSubscription {
    topic: Topic,
    id: UncheckedActivityId,
    closure: Handler,
    type_name: DebugTypeName,
}

impl NewSubscription {
    fn new(
        topic: Topic,
        id: impl Into<UncheckedActivityId>,
        closure: Handler,
        type_name: DebugTypeName,
    ) -> Self {
        Self {
            topic,
            id: id.into(),
            closure,
            type_name,
        }
    }
}

#[cfg(debug_assertions)]
impl std::fmt::Debug for NewSubscription {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        write!(
            f,
            "Adding new subscription to activity of type {:?}",
            self.type_name
        )
    }
}