mdebray/tvl-depot
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

refactor(main): Simplify receiver

Browse source 
Departing from the initial approach. There's no reason to multithread
this for now.
This commit is contained in:
Vincent Ambo 2018-05-28 09:45:07 +02:00
parent 6793b25a67
commit 0db4512df4
1 changed files with 8 additions and 21 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

29
src/main.rs
View file

@ -6,8 +6,6 @@ extern crate systemd;
use systemd::journal::*; use systemd::journal::*;
use std::process; use std::process;
use std::thread;
use std::sync::mpsc::{channel, Receiver};
use std::time::{Duration, Instant}; use std::time::{Duration, Instant};
use std::collections::vec_deque::{VecDeque, Drain}; use std::collections::vec_deque::{VecDeque, Drain};
@ -43,10 +41,11 @@ impl From<JournalRecord> for Record {
} }
} }
/// This function spawns a double-looped, blocking receiver. It will /// This function starts a double-looped, blocking receiver. It will
/// buffer messages for a second before flushing them to Stackdriver. /// buffer messages for half a second before flushing them to
fn receiver_loop(rx: Receiver<Record>) { /// Stackdriver.
let mut buf = VecDeque::new(); fn receiver_loop(mut journal: Journal) {
let mut buf: VecDeque<Record> = VecDeque::new();
let iteration = Duration::from_millis(500); let iteration = Duration::from_millis(500);
loop { loop {
@ -58,8 +57,8 @@ fn receiver_loop(rx: Receiver<Record>) {
break; break;
} }
if let Ok(record) = rx.recv_timeout(iteration) { if let Ok(Some(record)) = journal.await_next_record(Some(iteration)) {
buf.push_back(record); buf.push_back(record.into());
} }
} }
@ -91,17 +90,5 @@ fn main () {
} }
} }
let (tx, rx) = channel(); receiver_loop(journal)
let _receiver = thread::spawn(move || receiver_loop(rx));
journal.watch_all_elements(move |record| {
let record: Record = record.into();
if record.message.is_some() {
tx.send(record).ok();
}
Ok(())
}).expect("Failed to read new entries from journal");
} }