Allow the historian to track ownership of any type of data
This commit is contained in:
@ -6,7 +6,7 @@ use std::thread::sleep;
|
|||||||
use std::time::Duration;
|
use std::time::Duration;
|
||||||
use std::sync::mpsc::SendError;
|
use std::sync::mpsc::SendError;
|
||||||
|
|
||||||
fn create_log(hist: &Historian) -> Result<(), SendError<Event<Sha256Hash>>> {
|
fn create_log(hist: &Historian<Sha256Hash>) -> Result<(), SendError<Event<Sha256Hash>>> {
|
||||||
sleep(Duration::from_millis(15));
|
sleep(Duration::from_millis(15));
|
||||||
let data = Sha256Hash::default();
|
let data = Sha256Hash::default();
|
||||||
hist.sender.send(Event::Discovery { data })?;
|
hist.sender.send(Event::Discovery { data })?;
|
||||||
|
@ -9,11 +9,12 @@ use std::thread::JoinHandle;
|
|||||||
use std::sync::mpsc::{Receiver, Sender};
|
use std::sync::mpsc::{Receiver, Sender};
|
||||||
use std::time::{Duration, SystemTime};
|
use std::time::{Duration, SystemTime};
|
||||||
use log::{hash, hash_event, Entry, Event, Sha256Hash};
|
use log::{hash, hash_event, Entry, Event, Sha256Hash};
|
||||||
|
use serde::Serialize;
|
||||||
|
|
||||||
pub struct Historian {
|
pub struct Historian<T> {
|
||||||
pub sender: Sender<Event<Sha256Hash>>,
|
pub sender: Sender<Event<T>>,
|
||||||
pub receiver: Receiver<Entry<Sha256Hash>>,
|
pub receiver: Receiver<Entry<T>>,
|
||||||
pub thread_hdl: JoinHandle<(Entry<Sha256Hash>, ExitReason)>,
|
pub thread_hdl: JoinHandle<(Entry<T>, ExitReason)>,
|
||||||
}
|
}
|
||||||
|
|
||||||
#[derive(Debug, PartialEq, Eq)]
|
#[derive(Debug, PartialEq, Eq)]
|
||||||
@ -21,12 +22,12 @@ pub enum ExitReason {
|
|||||||
RecvDisconnected,
|
RecvDisconnected,
|
||||||
SendDisconnected,
|
SendDisconnected,
|
||||||
}
|
}
|
||||||
fn log_event(
|
fn log_event<T: Serialize + Clone>(
|
||||||
sender: &Sender<Entry<Sha256Hash>>,
|
sender: &Sender<Entry<T>>,
|
||||||
num_hashes: &mut u64,
|
num_hashes: &mut u64,
|
||||||
end_hash: &mut Sha256Hash,
|
end_hash: &mut Sha256Hash,
|
||||||
event: Event<Sha256Hash>,
|
event: Event<T>,
|
||||||
) -> Result<(), (Entry<Sha256Hash>, ExitReason)> {
|
) -> Result<(), (Entry<T>, ExitReason)> {
|
||||||
*end_hash = hash_event(end_hash, &event);
|
*end_hash = hash_event(end_hash, &event);
|
||||||
let entry = Entry {
|
let entry = Entry {
|
||||||
end_hash: *end_hash,
|
end_hash: *end_hash,
|
||||||
@ -40,15 +41,15 @@ fn log_event(
|
|||||||
Ok(())
|
Ok(())
|
||||||
}
|
}
|
||||||
|
|
||||||
fn log_events(
|
fn log_events<T: Serialize + Clone>(
|
||||||
receiver: &Receiver<Event<Sha256Hash>>,
|
receiver: &Receiver<Event<T>>,
|
||||||
sender: &Sender<Entry<Sha256Hash>>,
|
sender: &Sender<Entry<T>>,
|
||||||
num_hashes: &mut u64,
|
num_hashes: &mut u64,
|
||||||
end_hash: &mut Sha256Hash,
|
end_hash: &mut Sha256Hash,
|
||||||
epoch: SystemTime,
|
epoch: SystemTime,
|
||||||
num_ticks: &mut u64,
|
num_ticks: &mut u64,
|
||||||
ms_per_tick: Option<u64>,
|
ms_per_tick: Option<u64>,
|
||||||
) -> Result<(), (Entry<Sha256Hash>, ExitReason)> {
|
) -> Result<(), (Entry<T>, ExitReason)> {
|
||||||
use std::sync::mpsc::TryRecvError;
|
use std::sync::mpsc::TryRecvError;
|
||||||
loop {
|
loop {
|
||||||
if let Some(ms) = ms_per_tick {
|
if let Some(ms) = ms_per_tick {
|
||||||
@ -79,12 +80,12 @@ fn log_events(
|
|||||||
|
|
||||||
/// A background thread that will continue tagging received Event messages and
|
/// A background thread that will continue tagging received Event messages and
|
||||||
/// sending back Entry messages until either the receiver or sender channel is closed.
|
/// sending back Entry messages until either the receiver or sender channel is closed.
|
||||||
pub fn create_logger(
|
pub fn create_logger<T: 'static + Serialize + Clone + Send>(
|
||||||
start_hash: Sha256Hash,
|
start_hash: Sha256Hash,
|
||||||
ms_per_tick: Option<u64>,
|
ms_per_tick: Option<u64>,
|
||||||
receiver: Receiver<Event<Sha256Hash>>,
|
receiver: Receiver<Event<T>>,
|
||||||
sender: Sender<Entry<Sha256Hash>>,
|
sender: Sender<Entry<T>>,
|
||||||
) -> JoinHandle<(Entry<Sha256Hash>, ExitReason)> {
|
) -> JoinHandle<(Entry<T>, ExitReason)> {
|
||||||
use std::thread;
|
use std::thread;
|
||||||
thread::spawn(move || {
|
thread::spawn(move || {
|
||||||
let mut end_hash = start_hash;
|
let mut end_hash = start_hash;
|
||||||
@ -109,7 +110,7 @@ pub fn create_logger(
|
|||||||
})
|
})
|
||||||
}
|
}
|
||||||
|
|
||||||
impl Historian {
|
impl<T: 'static + Serialize + Clone + Send> Historian<T> {
|
||||||
pub fn new(start_hash: &Sha256Hash, ms_per_tick: Option<u64>) -> Self {
|
pub fn new(start_hash: &Sha256Hash, ms_per_tick: Option<u64>) -> Self {
|
||||||
use std::sync::mpsc::channel;
|
use std::sync::mpsc::channel;
|
||||||
let (sender, event_receiver) = channel();
|
let (sender, event_receiver) = channel();
|
||||||
@ -157,7 +158,7 @@ mod tests {
|
|||||||
#[test]
|
#[test]
|
||||||
fn test_historian_closed_sender() {
|
fn test_historian_closed_sender() {
|
||||||
let zero = Sha256Hash::default();
|
let zero = Sha256Hash::default();
|
||||||
let hist = Historian::new(&zero, None);
|
let hist = Historian::<u8>::new(&zero, None);
|
||||||
drop(hist.receiver);
|
drop(hist.receiver);
|
||||||
hist.sender.send(Event::Tick).unwrap();
|
hist.sender.send(Event::Tick).unwrap();
|
||||||
assert_eq!(
|
assert_eq!(
|
||||||
|
Reference in New Issue
Block a user