yozefu-app 0.0.30

The Kafka consumer of Yozefu
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139

use lib::{
    kafka::Comparable,
    search::{
        compare::{CompareExpression, NumberOperator, StringOperator},
        filter::Filter,
        offset::FromOffset,
    },
};

use super::SearchContext;
use crate::search::Search;

impl Search for CompareExpression {
    fn offset(&self) -> Option<FromOffset> {
        match self {
            CompareExpression::Offset(NumberOperator::Equal, e) => Some(FromOffset::Offset(*e)),
            CompareExpression::Offset(NumberOperator::GreaterOrEqual, e) => {
                Some(FromOffset::Offset(*e))
            }
            CompareExpression::Offset(NumberOperator::GreaterThan, e) => {
                Some(FromOffset::Offset(*e + 1))
            }
            CompareExpression::OffsetTail(e) => Some(FromOffset::OffsetTail(*e)),
            CompareExpression::Timestamp(op, e) => match op {
                NumberOperator::GreaterThan => Some(FromOffset::Timestamp(e.timestamp_millis())),
                NumberOperator::GreaterOrEqual => {
                    Some(FromOffset::Timestamp(e.timestamp_millis() - 1000))
                }
                NumberOperator::Equal => Some(FromOffset::Timestamp(e.timestamp_millis() - 1000)),
                NumberOperator::LowerThan => None,
                NumberOperator::LowerOrEqual => None,
                NumberOperator::NotEqual => None,
            },
            _ => None,
        }
    }

    fn matches(&self, context: &SearchContext) -> bool {
        let record = context.record;
        match self {
            CompareExpression::OffsetTail(_) => true,
            CompareExpression::Partition(op, p) => match op {
                &NumberOperator::GreaterThan => record.partition > *p,
                NumberOperator::GreaterOrEqual => record.partition >= *p,
                NumberOperator::LowerThan => record.partition < *p,
                NumberOperator::LowerOrEqual => record.partition <= *p,
                NumberOperator::Equal => record.partition == *p,
                NumberOperator::NotEqual => record.partition != *p,
            },
            CompareExpression::Offset(op, p) => match op {
                NumberOperator::GreaterThan => record.offset > *p,
                NumberOperator::GreaterOrEqual => record.offset >= *p,
                NumberOperator::LowerThan => record.offset < *p,
                NumberOperator::LowerOrEqual => record.offset <= *p,
                NumberOperator::Equal => record.offset == *p,
                NumberOperator::NotEqual => record.offset != *p,
            },
            CompareExpression::Topic(op, t) => match op {
                StringOperator::Equal => record.topic == *t,
                StringOperator::NotEqual => record.topic != *t,
                StringOperator::Contain => record.topic.contains(t),
                StringOperator::StartWith => record.topic.starts_with(t),
            },
            CompareExpression::Size(op, s) => match op {
                NumberOperator::GreaterThan => record.size > *s,
                NumberOperator::GreaterOrEqual => record.size >= *s,
                NumberOperator::LowerThan => record.size < *s,
                NumberOperator::LowerOrEqual => record.size <= *s,
                NumberOperator::Equal => record.size == *s,
                NumberOperator::NotEqual => record.size != *s,
            },
            CompareExpression::Key(op, t) => record.key.compare(&None, op, t),
            CompareExpression::Value(left, op, t) => record.value.compare(left, op, t),
            CompareExpression::Header(left, op, t) => {
                let headers = &record.headers;
                let header = headers.get(left);
                if header.is_none() {
                    return false;
                }
                let header = header.unwrap();
                match op {
                    StringOperator::Contain => header.contains(t),
                    StringOperator::Equal => header == t,
                    StringOperator::StartWith => header.starts_with(t),
                    StringOperator::NotEqual => header != t,
                }
            }
            CompareExpression::Timestamp(op, t) => {
                let ts = record.timestamp_as_local_date_time().unwrap();
                match op {
                    NumberOperator::GreaterThan => ts > *t,
                    NumberOperator::GreaterOrEqual => ts >= *t,
                    NumberOperator::LowerThan => ts < *t,
                    NumberOperator::LowerOrEqual => ts <= *t,
                    NumberOperator::Equal => ts == *t,
                    NumberOperator::NotEqual => ts != *t,
                }
            }
            CompareExpression::TimestampBetween(from, to) => {
                let ts = record.timestamp_as_local_date_time().unwrap();
                from <= &ts && &ts <= to
            }
        }
    }

    fn filters(&self) -> Vec<Filter> {
        vec![]
    }
}

#[test]
fn test_matches() {
    use crate::search::filter::CACHED_FILTERS;
    use lib::kafka::KafkaRecord;
    use std::path::PathBuf;

    let compare = CompareExpression::Offset(NumberOperator::Equal, 42);
    let record = KafkaRecord {
        topic: "test-topic".to_string(),
        partition: 0,
        offset: 42,
        key: lib::DataType::String("key".to_string()),
        value: lib::DataType::String("value".to_string()),
        timestamp: None,
        headers: std::collections::BTreeMap::new(),
        key_schema: None,
        value_schema: None,
        size: 12,
        key_as_string: "key".to_string(),
        value_as_string: "value".to_string(),
    };
    let context = SearchContext {
        record: &record,
        filters: &CACHED_FILTERS,
        filters_directory: PathBuf::from("."),
    };

    assert!(compare.matches(&context))
}