20 datafusion of all three topics into the report #24
Conversation
merger/src/consumer.rs
Outdated
| } | ||
|
|
||
| fn detect_event_type_topic2(payload: &Value) -> EventType { | ||
| if payload.get("truck_id").is_some() && payload.get("latitude").is_some() { |
There was a problem hiding this comment.
Instead of checking for the existence of two attributes, could you try to instantiating a Position object and in case of an error you return a EventType::Unknown.
There was a problem hiding this comment.
Or can you just get the "type" field and return the correct type ? E.g. :
match payload.get("type") {
Some(type) => match ...
None => EventType::Unknown
}There was a problem hiding this comment.
The issue is, we do not have a field that is present in all of the 4 eventType. "type" is present for driver and truck, but not for the other two.
There was a problem hiding this comment.
The field "type_" : "position" will be added to the Position element, to be able to use this method.
The types can be as followed : driver, truck, start, end, rest and position
| true | ||
| } else { | ||
| error!("Kafka is unreachable !"); | ||
| error!("Kafka is unreachable!"); |
There was a problem hiding this comment.
Please use the tracing crate instead of the standard one. Tracing provides great features such as log formatting (for json) or a loki exporter which are mandatory in a cloud environment
merger/src/consumer.rs
Outdated
| // List of topics to subscribe to | ||
| let topics = vec!["topic1", "topic2", "topic3"]; | ||
| // Shared HashMap for storing parsed messages | ||
| let shared_store: Arc<Mutex<HashMap<String, Vec<Value>>>> = Arc::new(Mutex::new(HashMap::new())); |
There was a problem hiding this comment.
Wouldn't it be better to reconstruct the objects and store the messages in different hashmaps ?
There was a problem hiding this comment.
Yes this is the end goal, I am not here yet
merger/src/consumer.rs
Outdated
| let shared_store = Arc::clone(&shared_store); | ||
| let topic = topic.clone(); | ||
|
|
||
| thread::spawn(move || { |
There was a problem hiding this comment.
Use tokio spawn instead of thread spawn. Actually tokio provides you a managed thread pool so you don't have to manage it yourself. Since you will be doing a lot of IO bound operations you should reallly use tokio since it was made for that
| consumer | ||
| .subscribe(&[&topic]) | ||
| .expect("Subscription to topic failed"); | ||
|
|
||
| loop { |
There was a problem hiding this comment.
If we want to do it the clean way, you should implement a graceful shutdown that listen for the ctrl+c unix event (interrupt). You can do that with a channel like in go https://doc.rust-lang.org/rust-by-example/std_misc/channels.html
No description provided.