Remove dynamic dispatch everywhere

Author: archis-polyverse

Cargo.toml

@@ -26,7 +26,7 @@ default = ["sync"]
 # packages, but they are strictly optional. Note that `session` is not a package
 # but rather another feature listed in this manifest.
 sync = []
-async = ["futures", "futures-util", "tokio"]
+async = ["futures", "futures-util", "tokio", "pin-project"]
 extra-traits = ["serde"]
 
 [dependencies]
@@ -51,11 +51,19 @@ serde = { version = "1.0.118", features = ["derive"], optional = true }
 futures = { version = "0.3.8", optional = true }
 futures-util = { version = "0.3.8", optional = true }
 tokio = { version = "1.0.1", features = ["rt", "fs", "io-util", "macros"], optional = true }
+pin-project = {version = "1.0.2", optional = true }
 
 [dev-dependencies]
 tokio-stream = { version = "0.1.0" }
 
-# Enable LTO for release (since it only builds in CI/CD and doesn't block day to day)
+[profile.dev]
+# We don't need stack unwinding in dev either - can be manually enabled
+panic = 'abort'
+
 [profile.release]
+# We don't need stack unwinding in releases
+panic = 'abort'
+# Enable LTO for release (since it only builds in Travis and doesn't block day to day)
 lto = "fat"
+# One code-gen unit so we get a highly optimized binary
 codegen-units = 1

src/entry.rs

@@ -1,9 +1,10 @@
 // Copyright (c) 2019 Polyverse Corporation
 
+use num_derive::FromPrimitive;
 use std::error::Error;
 use std::fmt::{Display, Formatter, Result as FmtResult};
 use std::time::Duration;
-use strum_macros::EnumString;
+use strum_macros::{Display, EnumString};
 
 #[cfg(feature = "extra-traits")]
 use serde::{Deserialize, Serialize};

src/klogctl.rs

@@ -12,11 +12,12 @@ use crate::entry::{Entry, EntryParsingError};
 use crate::error::RMesgError;
 
 use errno::errno;
+use lazy_static::lazy_static;
 use regex::Regex;
 use std::convert::TryFrom;
-use std::fmt::Display;
 use std::fs;
 use std::time::{Duration, SystemTime};
+use strum_macros::Display;
 
 /// suggest polling every ten seconds
 pub const SUGGESTED_POLL_INTERVAL: std::time::Duration = Duration::from_secs(10);

src/kmsgfile.rs

@@ -11,6 +11,7 @@ use crate::entry::{Entry, EntryParsingError};
 ///
 use crate::error::RMesgError;
 
+use lazy_static::lazy_static;
 use nonblock::NonBlockingReader;
 use regex::Regex;
 use std::fs as stdfs;

src/lib.rs

@@ -1,18 +1,9 @@
-#[macro_use]
-extern crate lazy_static;
-#[macro_use]
-extern crate enum_display_derive;
-#[macro_use]
-extern crate num_derive;
-
 mod common;
 
 pub mod entry;
 pub mod error;
-
 /// KLog Implementation (makes klogctl aka syslog system call through libc)
 pub mod klogctl;
-
 /// KMsg Implementation (reads from the /dev/kmsg file)
 pub mod kmsgfile;
 
@@ -23,6 +14,10 @@ use std::iter::Iterator;
 use core::pin::Pin;
 #[cfg(feature = "async")]
 use futures::stream::Stream;
+#[cfg(feature = "async")]
+use futures::task::{Context, Poll};
+#[cfg(feature = "async")]
+use pin_project::pin_project;
 
 #[derive(Clone, Copy, Debug)]
 pub enum Backend {
@@ -32,10 +27,37 @@ pub enum Backend {
 }
 
 #[cfg(feature = "sync")]
-pub type EntriesIterator = Box<dyn Iterator<Item = Result<entry::Entry, error::RMesgError>>>;
+pub enum EntriesIterator {
+    KLogCtl(klogctl::KLogEntries),
+    DevKMsg(kmsgfile::KMsgEntriesIter),
+}
+#[cfg(feature = "sync")]
+impl Iterator for EntriesIterator {
+    type Item = Result<entry::Entry, error::RMesgError>;
+    fn next(&mut self) -> Option<Self::Item> {
+        match self {
+            Self::KLogCtl(k) => k.next(),
+            Self::DevKMsg(d) => d.next(),
+        }
+    }
+}
 
+#[pin_project(project = EntriesStreamPinnedProjection)]
+#[cfg(feature = "async")]
+pub enum EntriesStream {
+    KLogCtl(#[pin] klogctl::KLogEntries),
+    DevKMsg(#[pin] kmsgfile::KMsgEntriesStream),
+}
 #[cfg(feature = "async")]
-pub type EntriesStream = Pin<Box<dyn Stream<Item = Result<entry::Entry, error::RMesgError>>>>;
+impl Stream for EntriesStream {
+    type Item = Result<entry::Entry, error::RMesgError>;
+    fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
+        match self.project() {
+            EntriesStreamPinnedProjection::KLogCtl(k) => k.poll_next(cx),
+            EntriesStreamPinnedProjection::DevKMsg(d) => d.poll_next(cx),
+        }
+    }
+}
 
 pub fn log_entries(b: Backend, clear: bool) -> Result<Vec<entry::Entry>, error::RMesgError> {
     match b {
@@ -77,20 +99,24 @@ pub fn logs_raw(b: Backend, clear: bool) -> Result<String, error::RMesgError> {
 pub fn logs_iter(b: Backend, clear: bool, raw: bool) -> Result<EntriesIterator, error::RMesgError> {
     match b {
         Backend::Default => match kmsgfile::KMsgEntriesIter::with_options(None, raw) {
-            Ok(e) => Ok(Box::new(e)),
+            Ok(e) => Ok(EntriesIterator::DevKMsg(e)),
             Err(error::RMesgError::DevKMsgFileOpenError(s)) => {
                 eprintln!(
                     "Falling back from device file to klogctl syscall due to error: {}",
                     s
                 );
-                Ok(Box::new(klog_entries_only_if_timestamp_enabled(clear)?))
+                Ok(EntriesIterator::KLogCtl(
+                    klog_entries_only_if_timestamp_enabled(clear)?,
+                ))
             }
             Err(e) => Err(e),
         },
-        Backend::KLogCtl => Ok(Box::new(klog_entries_only_if_timestamp_enabled(clear)?)),
-        Backend::DevKMsg => Ok(Box::new(kmsgfile::KMsgEntriesIter::with_options(
-            None, raw,
-        )?)),
+        Backend::KLogCtl => Ok(EntriesIterator::KLogCtl(
+            klog_entries_only_if_timestamp_enabled(clear)?,
+        )),
+        Backend::DevKMsg => Ok(EntriesIterator::DevKMsg(
+            kmsgfile::KMsgEntriesIter::with_options(None, raw)?,
+        )),
     }
 }
 
@@ -102,18 +128,22 @@ pub async fn logs_stream(
 ) -> Result<EntriesStream, error::RMesgError> {
     match b {
         Backend::Default => match kmsgfile::KMsgEntriesStream::with_options(None, raw).await {
-            Ok(e) => Ok(Box::pin(e)),
+            Ok(e) => Ok(EntriesStream::DevKMsg(e)),
             Err(error::RMesgError::DevKMsgFileOpenError(s)) => {
                 eprintln!(
                     "Falling back from device file to klogctl syscall due to error: {}",
                     s
                 );
-                Ok(Box::pin(klog_entries_only_if_timestamp_enabled(clear)?))
+                Ok(EntriesStream::KLogCtl(
+                    klog_entries_only_if_timestamp_enabled(clear)?,
+                ))
             }
             Err(e) => Err(e),
         },
-        Backend::KLogCtl => Ok(Box::pin(klog_entries_only_if_timestamp_enabled(clear)?)),
-        Backend::DevKMsg => Ok(Box::pin(
+        Backend::KLogCtl => Ok(EntriesStream::KLogCtl(
+            klog_entries_only_if_timestamp_enabled(clear)?,
+        )),
+        Backend::DevKMsg => Ok(EntriesStream::DevKMsg(
             kmsgfile::KMsgEntriesStream::with_options(None, raw).await?,
         )),
     }
@@ -135,3 +165,66 @@ fn klog_entries_only_if_timestamp_enabled(
 
     klogctl::KLogEntries::with_options(clear, klogctl::SUGGESTED_POLL_INTERVAL)
 }
+
+/**********************************************************************************/
+// Tests! Tests! Tests!
+
+#[cfg(all(test, target_os = "linux"))]
+mod test {
+    use super::*;
+    #[cfg(feature = "async")]
+    use tokio_stream::StreamExt;
+
+    #[test]
+    fn test_log_entries() {
+        let entries = log_entries(Backend::Default, false);
+        assert!(entries.is_ok(), "Response from kmsg not Ok");
+        assert!(!entries.unwrap().is_empty(), "Should have non-empty logs");
+    }
+
+    #[cfg(feature = "sync")]
+    #[test]
+    fn test_iterator() {
+        // uncomment below if you want to be extra-sure
+        //let enable_timestamp_result = kernel_log_timestamps_enable(true);
+        //assert!(enable_timestamp_result.is_ok());
+
+        // Don't clear the buffer. Poll every second.
+        let iterator_result = logs_iter(Backend::Default, false, false);
+        assert!(iterator_result.is_ok());
+
+        let iterator = iterator_result.unwrap();
+
+        // Read 10 lines and quit
+        for (count, entry) in iterator.enumerate() {
+            assert!(entry.is_ok());
+            if count > 10 {
+                break;
+            }
+        }
+    }
+
+    #[cfg(feature = "async")]
+    #[tokio::test]
+    async fn test_stream() {
+        // uncomment below if you want to be extra-sure
+        //let enable_timestamp_result = kernel_log_timestamps_enable(true);
+        //assert!(enable_timestamp_result.is_ok());
+
+        // Don't clear the buffer. Poll every second.
+        let stream_result = logs_stream(Backend::Default, false, false).await;
+        assert!(stream_result.is_ok());
+
+        let mut stream = stream_result.unwrap();
+
+        // Read 10 lines and quit
+        let mut count: u32 = 0;
+        while let Some(entry) = stream.next().await {
+            assert!(entry.is_ok());
+            count += 1;
+            if count > 10 {
+                break;
+            }
+        }
+    }
+}