v_module_queue/
veda_module.rs1use crate::module::{init_log, Module, PrepareError};
2use crate::signals::sys_sig_listener;
3use crossbeam_channel::{select, tick};
4use exitcode;
5use nng::{Protocol, Socket};
6use std::time::{Duration, Instant};
7use std::{thread, time};
8use v_individual_model::onto::individual::{Individual, RawObj};
9use v_individual_model::onto::parser::parse_raw;
10use v_queue::consumer::Consumer;
11use v_queue::record::ErrorQueue;
12
13pub trait VedaQueueModule {
14 fn before_batch(&mut self, size_batch: u32) -> Option<u32>;
15 fn prepare(&mut self, queue_element: &mut Individual) -> Result<bool, PrepareError>;
16 fn after_batch(&mut self, prepared_batch_size: u32) -> Result<bool, PrepareError>;
17 fn heartbeat(&mut self) -> Result<(), PrepareError>;
18 fn before_start(&mut self);
19 fn before_exit(&mut self);
20}
21
22impl Module {
23 pub fn prepare_queue(&mut self, veda_module: &mut dyn VedaQueueModule) {
24 init_log(&self.name);
25
26 let queue_consumer =
27 &mut Consumer::new("./data/queue", &self.name, "individuals-flow")
28 .expect("!!!!!!!!! FAIL QUEUE");
29
30 if let Ok(ch) = sys_sig_listener() {
31 self.syssig_ch = Some(ch);
32 }
33
34 let mut soc = Socket::new(Protocol::Sub0).unwrap();
35 let mut count_timeout_error = 0;
36
37 let mut prev_batch_time = Instant::now();
38 let update = tick(Duration::from_millis(1));
39 veda_module.before_start();
40 loop {
41 if let Some(qq) = &self.syssig_ch {
42 select! {
43 recv(update) -> _ => {
44 }
45 recv(qq) -> _ => {
46 info!("Exit");
47 veda_module.before_exit();
48 std::process::exit(exitcode::OK);
49 }
50 }
51 }
52
53 if let Err(PrepareError::Fatal) = veda_module.heartbeat() {
54 error!("heartbeat: found fatal error, stop listen queue");
55 break;
56 }
57
58 if let Some(s) = self.connect_to_notify_channel() {
59 soc = s;
60 }
61
62 if !self.get_info_of_part_with_retry(queue_consumer) {
63 continue;
64 }
65
66 let size_batch = queue_consumer.get_batch_size();
67
68 let mut max_size_batch = size_batch;
69 if let Some(m) = self.max_batch_size {
70 max_size_batch = m;
71 }
72
73 if size_batch > 0 {
74 debug!("queue: batch size={}", size_batch);
75 if let Some(new_size) = veda_module.before_batch(size_batch) {
76 max_size_batch = new_size;
77 }
78 }
79
80 let mut prepared_batch_size = 0;
81 for _it in 0..max_size_batch {
82 if !queue_consumer.pop_header() {
83 break;
84 }
85
86 let mut raw = RawObj::new(vec![0; queue_consumer.record_len()]);
87
88 if let Err(e) = queue_consumer.pop_body(&mut raw.data) {
89 match e {
90 ErrorQueue::FailReadTailMessage => break,
91 ErrorQueue::InvalidChecksum => {
92 error!("[module] consumer:pop_body: invalid CRC, attempt seek next record");
93 queue_consumer.seek_next_pos();
94 break;
95 }
96 _ => {
97 error!(
98 "{} get msg from queue: {}",
99 self.queue_prepared_count,
100 e.as_str()
101 );
102 break;
103 }
104 }
105 }
106
107 let mut need_commit = true;
108
109 let mut queue_element = Individual::new_raw(raw);
110 if parse_raw(&mut queue_element).is_ok() {
111 let mut is_processed = true;
112 if let Some(assigned_subsystems) =
113 queue_element.get_first_integer("assigned_subsystems")
114 {
115 if assigned_subsystems > 0 {
116 if let Some(my_subsystem_id) = self.subsystem_id {
117 if assigned_subsystems & my_subsystem_id == 0 {
118 is_processed = false;
119 }
120 } else {
121 is_processed = false;
122 }
123 }
124 }
125
126 if is_processed {
127 match veda_module.prepare(&mut queue_element) {
128 Err(e) => {
129 if let PrepareError::Fatal = e {
130 warn!("prepare: found fatal error, stop listen queue");
131 return;
132 }
133 }
134 Ok(b) => {
135 need_commit = b;
136 }
137 }
138 }
139 }
140
141 if need_commit {
142 queue_consumer.commit();
143 }
144
145 self.queue_prepared_count += 1;
146
147 if self.queue_prepared_count % 1000 == 0 {
148 info!("get from queue, count: {}", self.queue_prepared_count);
149 }
150 prepared_batch_size += 1;
151 }
152
153 if size_batch > 0 {
154 match veda_module.after_batch(prepared_batch_size) {
155 Ok(b) => {
156 if b {
157 queue_consumer.commit();
158 }
159 }
160 Err(e) => {
161 if let PrepareError::Fatal = e {
162 warn!("after_batch: found fatal error, stop listen queue");
163 return;
164 }
165 }
166 }
167 }
168
169 if prepared_batch_size == size_batch {
170 let wmsg = soc.recv();
171 if let Err(e) = wmsg {
172 debug!("fail recv from queue notify channel, err={:?}", e);
173
174 if count_timeout_error > 0 && size_batch > 0 {
175 warn!("queue changed but we not received notify message, need reconnect...");
176 self.is_ready_notify_channel = false;
177 count_timeout_error += 1;
178 }
179 } else {
180 count_timeout_error = 0;
181 }
182 }
183
184 if let Some(t) = self.max_timeout_between_batches {
185 let delta = prev_batch_time.elapsed().as_millis() as u64;
186 if let Some(c) = self.min_batch_size_to_cancel_timeout {
187 if prepared_batch_size < c && delta < t {
188 thread::sleep(time::Duration::from_millis(t - delta));
189 info!("sleep {} ms", t - delta);
190 }
191 } else if delta < t {
192 thread::sleep(time::Duration::from_millis(t - delta));
193 info!("sleep {} ms", t - delta);
194 }
195 }
196
197 prev_batch_time = Instant::now();
198 }
199 }
200}