use crate::{Topic, Partition, Offset};
use derive_more::{Deref, DerefMut, From, Into};
use serde::{Deserialize, Serialize};
use std::time::{SystemTime, UNIX_EPOCH};
use uuid::Uuid;
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize, Deref, DerefMut, From, Into)]
pub struct EventId(pub Uuid);
impl EventId {
pub fn new() -> Self {
Self(Uuid::new_v4())
}
pub fn from_uuid(uuid: Uuid) -> Self {
Self(uuid)
}
}
impl Default for EventId {
fn default() -> Self {
Self::new()
}
}
impl std::fmt::Display for EventId {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
write!(f, "{}", self.0)
}
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, Deref, DerefMut, From, Into)]
pub struct EventData(pub Vec<u8>);
impl EventData {
pub fn new(data: Vec<u8>) -> Self {
Self(data)
}
pub fn from_bytes(bytes: impl Into<Vec<u8>>) -> Self {
Self(bytes.into())
}
pub fn from_json(json: &str) -> Self {
Self(json.as_bytes().to_vec())
}
pub fn len(&self) -> usize {
self.0.len()
}
pub fn is_empty(&self) -> bool {
self.0.is_empty()
}
pub fn as_bytes(&self) -> &[u8] {
&self.0
}
pub fn as_str(&self) -> Option<&str> {
std::str::from_utf8(&self.0).ok()
}
pub fn to_json(&self) -> String {
match std::str::from_utf8(&self.0) {
Ok(s) => s.to_string(),
Err(_) => {
use base64::Engine;
base64::engine::general_purpose::STANDARD.encode(&self.0)
}
}
}
pub fn to_json_value(&self) -> Result<serde_json::Value, serde_json::Error> {
match std::str::from_utf8(&self.0) {
Ok(s) => serde_json::from_str(s),
Err(_) => {
use base64::Engine;
let base64_str = base64::engine::general_purpose::STANDARD.encode(&self.0);
Ok(serde_json::Value::String(base64_str))
}
}
}
pub fn from_json_validated(json: &str) -> Result<Self, serde_json::Error> {
let _: serde_json::Value = serde_json::from_str(json)?;
Ok(Self::from_json(json))
}
pub fn to_text(&self) -> String {
match std::str::from_utf8(&self.0) {
Ok(s) => s.to_string(),
Err(_) => format!("<binary data: {} bytes>", self.0.len()),
}
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize, Deref, DerefMut, From, Into)]
pub struct Timestamp(pub u64);
impl Timestamp {
pub fn now() -> Self {
let duration = SystemTime::now()
.duration_since(UNIX_EPOCH)
.unwrap_or_default();
Self(duration.as_millis() as u64)
}
pub fn from_millis(millis: u64) -> Self {
Self(millis)
}
pub fn as_millis(&self) -> u64 {
self.0
}
}
impl Default for Timestamp {
fn default() -> Self {
Self::now()
}
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct EventMetadata {
pub producer_id: Option<String>,
pub content_type: Option<String>,
pub event_type: Option<String>,
pub headers: std::collections::HashMap<String, String>,
}
impl Default for EventMetadata {
fn default() -> Self {
Self {
producer_id: None,
content_type: None,
event_type: None,
headers: std::collections::HashMap::new(),
}
}
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct Event {
pub id: EventId,
pub topic: Topic,
pub partition: Partition,
pub offset: Option<Offset>,
pub timestamp: Timestamp,
pub data: EventData,
pub metadata: EventMetadata,
}
impl Event {
pub fn new(
id: EventId,
topic: Topic,
partition: Partition,
data: EventData,
) -> Self {
Self {
id,
topic,
partition,
offset: None,
timestamp: Timestamp::now(),
data,
metadata: EventMetadata::default(),
}
}
pub fn with_metadata(
id: EventId,
topic: Topic,
partition: Partition,
data: EventData,
metadata: EventMetadata,
) -> Self {
Self {
id,
topic,
partition,
offset: None,
timestamp: Timestamp::now(),
data,
metadata,
}
}
pub fn set_offset(&mut self, offset: Offset) {
self.offset = Some(offset);
}
pub fn estimated_size(&self) -> usize {
std::mem::size_of::<EventId>()
+ self.topic.len()
+ std::mem::size_of::<Partition>()
+ std::mem::size_of::<Option<Offset>>()
+ std::mem::size_of::<Timestamp>()
+ self.data.len()
+ self.metadata.headers.len() * 50 }
pub fn sort_key(&self) -> (Timestamp, EventId) {
(self.timestamp, self.id)
}
}