pub mod fluentd;
#[cfg(feature = "kafka")]
pub mod kafka;
pub mod mbox;
#[cfg(feature = "ndarray")]
pub mod ndarray;
#[cfg(feature = "pcap")]
pub mod pcap;
mod pipeline;
pub mod text;
pub use pipeline::split;
use std::error;
use std::fmt;
pub trait Input {
type Data;
type Ack;
fn run(self) -> Result<(), Error>;
}
pub type SeqNo = usize;
pub trait Event {
type Ack;
fn raw(&self) -> &[u8];
fn time(&self) -> SeqNo;
fn ack(&self) -> Self::Ack;
}
#[derive(Debug)]
pub struct BareEvent {
pub raw: Vec<u8>,
pub seq_no: SeqNo,
}
impl Event for BareEvent {
type Ack = SeqNo;
#[must_use]
fn raw(&self) -> &[u8] {
self.raw.as_slice()
}
#[must_use]
fn time(&self) -> SeqNo {
self.seq_no
}
#[must_use]
fn ack(&self) -> Self::Ack {
self.seq_no
}
}
#[derive(Debug)]
pub enum Error {
ChannelClosed,
CannotCommit(Box<dyn error::Error>),
CannotFetch(Box<dyn error::Error>),
InvalidMessage(Box<dyn error::Error>),
Fatal(String),
TooManyEvents(usize),
}
impl error::Error for Error {
#[must_use]
fn source(&self) -> Option<&(dyn error::Error + 'static)> {
match self {
Self::CannotCommit(e) | Self::CannotFetch(e) | Self::InvalidMessage(e) => {
Some(e.as_ref())
}
_ => None,
}
}
}
impl fmt::Display for Error {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
match self {
Self::ChannelClosed => write!(f, "data channel closed"),
Self::CannotCommit(e) => write!(f, "cannot commit to Kafka: {e}"),
Self::CannotFetch(e) => write!(f, "cannot fetch message from Kafka: {e}"),
Self::InvalidMessage(e) => write!(f, "invalid MessagePack format: {e}"),
Self::Fatal(s) => write!(f, "fatal error: {s}"),
Self::TooManyEvents(n) => {
write!(f, "cannot handle {n} events (expected < {})", u32::MAX)
}
}
}
}