splinter 0.3.14

Splinter is a privacy-focused platform for distributed applications that provides a blockchain-inspired networking environment for communication and transactions between organizations.
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

// Copyright 2018-2020 Cargill Incorporated
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

use std;
use std::sync::{
    atomic::{AtomicBool, Ordering},
    mpsc::Sender,
    Arc,
};
use std::thread;
use std::time::Duration;

use crate::network::connection_manager::error::ConnectionManagerError;

pub struct Pacemaker {
    interval: u64,
    join_handle: Option<thread::JoinHandle<()>>,
    shutdown_handle: Option<ShutdownHandle>,
}

impl Pacemaker {
    pub fn new(interval: u64) -> Self {
        Self {
            interval,
            join_handle: None,
            shutdown_handle: None,
        }
    }

    pub fn start<M, F>(
        &mut self,
        cm_sender: Sender<M>,
        new_message: F,
    ) -> Result<(), ConnectionManagerError>
    where
        M: Send + 'static,
        F: Fn() -> M + Send + 'static,
    {
        if self.join_handle.is_some() {
            return Ok(());
        }

        let running = Arc::new(AtomicBool::new(true));

        let running_clone = running.clone();
        let interval = self.interval;
        let join_handle = thread::Builder::new()
            .name("Heartbeat Monitor".into())
            .spawn(move || {
                info!("Starting heartbeat manager");

                while running_clone.load(Ordering::SeqCst) {
                    thread::sleep(Duration::from_secs(interval));
                    if let Err(err) = cm_sender.send(new_message()) {
                        error!(
                            "Connection manager has disconnected before 
                            shutting down heartbeat monitor {:?}",
                            err
                        );
                        break;
                    }
                }
            })?;

        self.join_handle = Some(join_handle);
        self.shutdown_handle = Some(ShutdownHandle { running });

        Ok(())
    }

    pub fn shutdown_handle(&self) -> Option<ShutdownHandle> {
        self.shutdown_handle.clone()
    }

    pub fn await_shutdown(self) {
        let join_handle = if let Some(jh) = self.join_handle {
            jh
        } else {
            return;
        };

        if let Err(err) = join_handle.join() {
            error!("Failed to shutdown heartbeat monitor gracefully: {:?}", err);
        }
    }
}

#[derive(Clone)]
pub struct ShutdownHandle {
    running: Arc<AtomicBool>,
}

impl ShutdownHandle {
    pub fn shutdown(&self) {
        self.running.store(false, Ordering::SeqCst)
    }
}