Allow the logger to inject Tick events on its own
This commit is contained in:
parent
d7dfa8c22d
commit
b34d2d7dee
|
@ -45,7 +45,7 @@ fn create_log(hist: &Historian) -> Result<(), SendError<Event>> {
|
||||||
|
|
||||||
fn main() {
|
fn main() {
|
||||||
let seed = Sha256Hash::default();
|
let seed = Sha256Hash::default();
|
||||||
let hist = Historian::new(&seed);
|
let hist = Historian::new(&seed, None);
|
||||||
create_log(&hist).expect("send error");
|
create_log(&hist).expect("send error");
|
||||||
drop(hist.sender);
|
drop(hist.sender);
|
||||||
let entries: Vec<Entry> = hist.receiver.iter().collect();
|
let entries: Vec<Entry> = hist.receiver.iter().collect();
|
||||||
|
|
|
@ -16,7 +16,7 @@ fn create_log(hist: &Historian) -> Result<(), SendError<Event>> {
|
||||||
|
|
||||||
fn main() {
|
fn main() {
|
||||||
let seed = Sha256Hash::default();
|
let seed = Sha256Hash::default();
|
||||||
let hist = Historian::new(&seed);
|
let hist = Historian::new(&seed, None);
|
||||||
create_log(&hist).expect("send error");
|
create_log(&hist).expect("send error");
|
||||||
drop(hist.sender);
|
drop(hist.sender);
|
||||||
let entries: Vec<Entry> = hist.receiver.iter().collect();
|
let entries: Vec<Entry> = hist.receiver.iter().collect();
|
||||||
|
|
|
@ -7,6 +7,7 @@
|
||||||
|
|
||||||
use std::thread::JoinHandle;
|
use std::thread::JoinHandle;
|
||||||
use std::sync::mpsc::{Receiver, Sender};
|
use std::sync::mpsc::{Receiver, Sender};
|
||||||
|
use std::time::{Duration, SystemTime};
|
||||||
use log::{extend_and_hash, hash, Entry, Event, Sha256Hash};
|
use log::{extend_and_hash, hash, Entry, Event, Sha256Hash};
|
||||||
|
|
||||||
pub struct Historian {
|
pub struct Historian {
|
||||||
|
@ -20,17 +21,12 @@ pub enum ExitReason {
|
||||||
RecvDisconnected,
|
RecvDisconnected,
|
||||||
SendDisconnected,
|
SendDisconnected,
|
||||||
}
|
}
|
||||||
|
fn log_event(
|
||||||
fn log_events(
|
|
||||||
receiver: &Receiver<Event>,
|
|
||||||
sender: &Sender<Entry>,
|
sender: &Sender<Entry>,
|
||||||
num_hashes: &mut u64,
|
num_hashes: &mut u64,
|
||||||
end_hash: &mut Sha256Hash,
|
end_hash: &mut Sha256Hash,
|
||||||
|
event: Event,
|
||||||
) -> Result<(), (Entry, ExitReason)> {
|
) -> Result<(), (Entry, ExitReason)> {
|
||||||
use std::sync::mpsc::TryRecvError;
|
|
||||||
loop {
|
|
||||||
match receiver.try_recv() {
|
|
||||||
Ok(event) => {
|
|
||||||
if let Event::UserDataKey(key) = event {
|
if let Event::UserDataKey(key) = event {
|
||||||
*end_hash = extend_and_hash(end_hash, &key);
|
*end_hash = extend_and_hash(end_hash, &key);
|
||||||
}
|
}
|
||||||
|
@ -43,6 +39,30 @@ fn log_events(
|
||||||
return Err((entry, ExitReason::SendDisconnected));
|
return Err((entry, ExitReason::SendDisconnected));
|
||||||
}
|
}
|
||||||
*num_hashes = 0;
|
*num_hashes = 0;
|
||||||
|
Ok(())
|
||||||
|
}
|
||||||
|
|
||||||
|
fn log_events(
|
||||||
|
receiver: &Receiver<Event>,
|
||||||
|
sender: &Sender<Entry>,
|
||||||
|
num_hashes: &mut u64,
|
||||||
|
end_hash: &mut Sha256Hash,
|
||||||
|
epoch: SystemTime,
|
||||||
|
num_ticks: &mut u64,
|
||||||
|
ms_per_tick: Option<u64>,
|
||||||
|
) -> Result<(), (Entry, ExitReason)> {
|
||||||
|
use std::sync::mpsc::TryRecvError;
|
||||||
|
loop {
|
||||||
|
if let Some(ms) = ms_per_tick {
|
||||||
|
let now = SystemTime::now();
|
||||||
|
if now > epoch + Duration::from_millis((*num_ticks + 1) * ms) {
|
||||||
|
log_event(sender, num_hashes, end_hash, Event::Tick)?;
|
||||||
|
*num_ticks += 1;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
match receiver.try_recv() {
|
||||||
|
Ok(event) => {
|
||||||
|
log_event(sender, num_hashes, end_hash, event)?;
|
||||||
}
|
}
|
||||||
Err(TryRecvError::Empty) => {
|
Err(TryRecvError::Empty) => {
|
||||||
return Ok(());
|
return Ok(());
|
||||||
|
@ -63,6 +83,7 @@ fn log_events(
|
||||||
/// 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(
|
||||||
start_hash: Sha256Hash,
|
start_hash: Sha256Hash,
|
||||||
|
ms_per_tick: Option<u64>,
|
||||||
receiver: Receiver<Event>,
|
receiver: Receiver<Event>,
|
||||||
sender: Sender<Entry>,
|
sender: Sender<Entry>,
|
||||||
) -> JoinHandle<(Entry, ExitReason)> {
|
) -> JoinHandle<(Entry, ExitReason)> {
|
||||||
|
@ -70,8 +91,18 @@ pub fn create_logger(
|
||||||
thread::spawn(move || {
|
thread::spawn(move || {
|
||||||
let mut end_hash = start_hash;
|
let mut end_hash = start_hash;
|
||||||
let mut num_hashes = 0;
|
let mut num_hashes = 0;
|
||||||
|
let mut num_ticks = 0;
|
||||||
|
let epoch = SystemTime::now();
|
||||||
loop {
|
loop {
|
||||||
if let Err(err) = log_events(&receiver, &sender, &mut num_hashes, &mut end_hash) {
|
if let Err(err) = log_events(
|
||||||
|
&receiver,
|
||||||
|
&sender,
|
||||||
|
&mut num_hashes,
|
||||||
|
&mut end_hash,
|
||||||
|
epoch,
|
||||||
|
&mut num_ticks,
|
||||||
|
ms_per_tick,
|
||||||
|
) {
|
||||||
return err;
|
return err;
|
||||||
}
|
}
|
||||||
end_hash = hash(&end_hash);
|
end_hash = hash(&end_hash);
|
||||||
|
@ -81,11 +112,11 @@ pub fn create_logger(
|
||||||
}
|
}
|
||||||
|
|
||||||
impl Historian {
|
impl Historian {
|
||||||
pub fn new(start_hash: &Sha256Hash) -> 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();
|
||||||
let (entry_sender, receiver) = channel();
|
let (entry_sender, receiver) = channel();
|
||||||
let thread_hdl = create_logger(*start_hash, event_receiver, entry_sender);
|
let thread_hdl = create_logger(*start_hash, ms_per_tick, event_receiver, entry_sender);
|
||||||
Historian {
|
Historian {
|
||||||
sender,
|
sender,
|
||||||
receiver,
|
receiver,
|
||||||
|
@ -105,7 +136,7 @@ mod tests {
|
||||||
use std::time::Duration;
|
use std::time::Duration;
|
||||||
|
|
||||||
let zero = Sha256Hash::default();
|
let zero = Sha256Hash::default();
|
||||||
let hist = Historian::new(&zero);
|
let hist = Historian::new(&zero, None);
|
||||||
|
|
||||||
hist.sender.send(Event::Tick).unwrap();
|
hist.sender.send(Event::Tick).unwrap();
|
||||||
sleep(Duration::new(0, 1_000_000));
|
sleep(Duration::new(0, 1_000_000));
|
||||||
|
@ -129,7 +160,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);
|
let hist = Historian::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!(
|
||||||
|
|
Loading…
Reference in New Issue