peermerge 0.0.3

Manage JSON-like documents with multiple writers, without a central authority, using a P2P protocol
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

use dashmap::DashMap;
use std::sync::Arc;

use crate::common::utils::{Mutex, YieldNow};
use crate::FeedDiscoveryKey;

use super::{Feed, FeedPersistence};

pub(crate) type FeedMap<U> = Arc<DashMap<FeedDiscoveryKey, Arc<Mutex<Feed<U>>>>>;
pub(crate) const SHUTDOWN_SIGNAL_NAME: &str = "shutdown";

pub(crate) async fn get_feed<U>(
    feeds: &FeedMap<U>,
    discovery_key: &FeedDiscoveryKey,
) -> Option<Arc<Mutex<Feed<U>>>>
where
    U: FeedPersistence,
{
    loop {
        match feeds.try_get(discovery_key) {
            dashmap::try_result::TryResult::Absent => {
                return None;
            }
            dashmap::try_result::TryResult::Locked => {
                YieldNow(false).await;
            }
            dashmap::try_result::TryResult::Present(value) => {
                return Some(value.clone());
            }
        }
    }
}

pub(crate) async fn insert_feed<U>(
    feeds: &FeedMap<U>,
    feed: Feed<U>,
    discovery_key: &FeedDiscoveryKey,
) -> bool
where
    U: FeedPersistence,
{
    loop {
        if let Some(entry) = feeds.try_entry(*discovery_key) {
            match entry {
                dashmap::mapref::entry::Entry::Occupied(_) => {
                    break false;
                }
                dashmap::mapref::entry::Entry::Vacant(vacant) => {
                    vacant.insert(Arc::new(Mutex::new(feed)));
                    break true;
                }
            }
        } else {
            YieldNow(false).await;
        }
    }
}

pub(crate) async fn remove_feed<U>(
    feeds: &FeedMap<U>,
    discovery_key: &FeedDiscoveryKey,
) -> Option<Arc<Mutex<Feed<U>>>>
where
    U: FeedPersistence,
{
    loop {
        if let Some(entry) = feeds.try_entry(*discovery_key) {
            match entry {
                dashmap::mapref::entry::Entry::Occupied(value) => {
                    break Some(value.remove());
                }
                dashmap::mapref::entry::Entry::Vacant(_) => {
                    break None;
                }
            }
        } else {
            YieldNow(false).await;
        }
    }
}

pub(crate) async fn get_feed_discovery_keys<U>(feeds: &FeedMap<U>) -> Vec<[u8; 32]>
where
    U: FeedPersistence,
{
    // I believe this needs to be resolved xacrimon/dashmap/issues/151 for this to guarantee
    // to not deadlock.
    feeds.iter().map(|multi| *multi.key()).collect()
}