dcs2 0.1.0

An extensible distributed control system framework made in rust with no-std support.
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

use log::info;

use crate::communication::messages::{IdentificableMessage, SimpleCodifier, Messages, ControlMessage};
use crate::communication::service::GenericCommunicationService;
use crate::membership::client::{MembershipClient, MembershipManager};
use crate::membership::metadata::{BasicMetadata, NodeMetadata};
use crate::nodes::CommunicationSDK;
use crate::properties::{EncodedMetadata, init_buffer, SystemBufferVec};

pub type SystemCommunicationService<CommsSDK> = GenericCommunicationService<
    <CommsSDK as CommunicationSDK>::Connection,
    <CommsSDK as CommunicationSDK>::Address,
    <CommsSDK as CommunicationSDK>::Factory,
    <CommsSDK as CommunicationSDK>::Router,
>;

pub fn connect_with_members<
    Msg: IdentificableMessage,
    CommsSDK: CommunicationSDK,
    CoordMetadata: BasicMetadata,
    Membership: MembershipManager<SIZE, NodeMetadata = NodeMetadata<CoordMetadata, CommsSDK::Metadata>>,
    MsgCod: SimpleCodifier<Data= Messages<Msg>>,
    const SIZE: usize,
>(
    metadata: EncodedMetadata,
    membership: &mut Membership,
    communication: &mut SystemCommunicationService<CommsSDK>,
    codifier: &mut MsgCod,
) where
    Membership: MembershipClient<Address = CommsSDK::Address>,
{
    info!("Connecting to network members.");
    let message: Messages<Msg> = Messages::System(ControlMessage::GetMetadata(metadata));

    let (removed_members, new_members) = membership.find_members();
    info!(
        "There are {} new members and {} removed members in the network.",
        new_members.len(),
        removed_members.len()
    );

    for removed_member in removed_members {
        if let Some(old_id) = communication.unpeer(removed_member) {
            membership.unregister_member(old_id);
        }
    }

    let mut buffer = init_buffer();
    let codified = codifier.encode(&message, &mut buffer).unwrap();

    if let Ok(bytes) = SystemBufferVec::from_slice(&buffer[..codified]) {
        for new_member in new_members {
            info!("Establishing handshake with node at {}.", new_member);
            communication.handshake(new_member, bytes.clone())
        }
    }
}