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
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

use futures::channel::mpsc::UnboundedSender;
use futures::stream::StreamExt;
use hypercore_protocol::{hypercore::Hypercore, Channel, ChannelReceiver, Message};
use random_access_storage::RandomAccess;
use std::fmt::Debug;
use std::sync::Arc;
use tracing::{debug, instrument};

use super::{
    messaging::{create_broadcast_message, create_initial_synchronize},
    on_message, PeerState,
};
use crate::{
    common::{utils::Mutex, FeedEvent, FeedEventContent},
    PeermergeError,
};

#[instrument(level = "debug", skip_all)]
pub(super) async fn on_feed<T>(
    mut hypercore: Arc<Mutex<Hypercore<T>>>,
    mut peer_state: PeerState,
    mut channel: Channel,
    mut channel_receiver: ChannelReceiver<Message>,
    feed_event_sender: &mut UnboundedSender<FeedEvent>,
) -> Result<(), PeermergeError>
where
    T: RandomAccess + Debug + Send + 'static,
{
    // Immediately send an initial synchronize
    let messages = create_initial_synchronize(&mut hypercore, &mut peer_state).await;
    channel.send_batch(&messages).await?;

    // Start listening on incoming messages or internal messages
    debug!("Start listening on channel messages");
    while let Some(message) = channel_receiver.next().await {
        let exit = process_message(
            message,
            &mut hypercore,
            &mut peer_state,
            &mut channel,
            feed_event_sender,
        )
        .await?;
        if exit {
            break;
        }
    }
    debug!("Exiting");
    Ok(())
}

#[instrument(level = "debug", skip_all)]
pub(super) async fn on_doc_feed<T>(
    mut hypercore: Arc<Mutex<Hypercore<T>>>,
    mut peer_state: PeerState,
    mut channel: Channel,
    mut channel_receiver: ChannelReceiver<Message>,
    feed_event_sender: &mut UnboundedSender<FeedEvent>,
) -> Result<(), PeermergeError>
where
    T: RandomAccess + Debug + Send + 'static,
{
    // Immediately broadcast hypercores to the other end on the doc peer, and
    // also start initial sync.
    let mut messages: Vec<Message> = vec![create_broadcast_message(
        peer_state.feeds_state.as_ref().unwrap(),
        &peer_state.child_documents,
        None,
    )];
    messages.extend(create_initial_synchronize(&mut hypercore, &mut peer_state).await);
    channel.send_batch(&messages).await?;

    // Start listening on incoming messages or internal messages
    debug!("Start listening on doc channel messages");
    while let Some(message) = channel_receiver.next().await {
        let exit = process_message(
            message,
            &mut hypercore,
            &mut peer_state,
            &mut channel,
            feed_event_sender,
        )
        .await?;
        if exit {
            break;
        }
    }
    debug!("Exiting");
    Ok(())
}

#[instrument(level = "debug", skip_all)]
async fn process_message<T>(
    message: Message,
    hypercore: &mut Arc<Mutex<Hypercore<T>>>,
    peer_state: &mut PeerState,
    channel: &mut Channel,
    feed_event_sender: &mut UnboundedSender<FeedEvent>,
) -> Result<bool, PeermergeError>
where
    T: RandomAccess + Debug + Send + 'static,
{
    let events = on_message(hypercore, peer_state, channel, message).await?;
    let mut disconnect_channel_id = None;
    for event in events {
        if let FeedEventContent::FeedDisconnected { channel } = event.content {
            disconnect_channel_id = Some(channel);
        }
        feed_event_sender.unbounded_send(event)?;
    }
    if let Some(id) = disconnect_channel_id {
        debug!("Disconnected channel {id}");
        return Ok(true);
    }
    Ok(false)
}