1use crate::info::ModuleInfo;
2use crate::signals::sys_sig_listener;
3use crate::api::IndvOp;
4use chrono::Local;
5use crossbeam_channel::{select, tick, Receiver};
6use env_logger::Builder;
7use ini::Ini;
8use nng::options::protocol::pubsub::Subscribe;
9use nng::options::Options;
10use nng::options::RecvTimeout;
11use nng::{Protocol, Socket};
12use std::io::Write;
13use std::str::FromStr;
14use std::time::Duration;
15use std::time::Instant;
16use std::{env, thread, time};
17use v_individual_model::onto::individual::{Individual, RawObj};
18use v_individual_model::onto::parser::parse_raw;
19use v_queue::consumer::*;
20use v_queue::record::*;
21
22#[derive(Debug)]
23#[repr(u8)]
24pub enum PrepareError {
25 Fatal = 101,
26 Recoverable = 102,
27}
28
29const NOTIFY_CHANNEL_RECONNECT_TIMEOUT: u64 = 300;
30const DEFAULT_QUEUE_PART_RETRY_COUNT: u32 = 5;
31const DEFAULT_QUEUE_PART_RETRY_PAUSE_MS: u64 = 1000;
32
33pub struct Module {
34 pub(crate) queue_prepared_count: i64,
35 notify_channel_url: String,
36 pub(crate) is_ready_notify_channel: bool,
37 notify_channel_read_timeout: Option<u64>,
38 pub(crate) max_timeout_between_batches: Option<u64>,
39 pub(crate) min_batch_size_to_cancel_timeout: Option<u32>,
40 pub max_batch_size: Option<u32>,
41 pub(crate) subsystem_id: Option<i64>,
42 pub(crate) syssig_ch: Option<Receiver<i32>>,
43 pub name: String,
44 pub onto_types: Vec<String>,
45 queue_part_retry_count: u32,
46 queue_part_retry_pause_ms: u64,
47}
48
49impl Default for Module {
50 fn default() -> Self {
51 Module::create(None, "")
52 }
53}
54
55impl Module {
56 pub fn new_with_name(name: &str) -> Self {
57 Module::create(None, name)
58 }
59
60 pub fn create(module_id: Option<i64>, module_name: &str) -> Self {
61 let args: Vec<String> = env::args().collect();
62
63 let mut notify_channel_url = String::default();
64 let mut max_timeout_between_batches = None;
65 let mut min_batch_size_to_cancel_timeout = None;
66 let mut max_batch_size = None;
67 let mut notify_channel_read_timeout = None;
68
69 for el in args.iter() {
70 if el.starts_with("--max_timeout_between_batches") {
71 let p: Vec<&str> = el.split('=').collect();
72 if let Ok(v) = p[1].parse::<u64>() {
73 max_timeout_between_batches = Some(v);
74 info!("use {} = {} ms", p[0], v);
75 }
76 } else if el.starts_with("--min_batch_size_to_cancel_timeout") {
77 let p: Vec<&str> = el.split('=').collect();
78 if let Ok(v) = p[1].parse::<u32>() {
79 min_batch_size_to_cancel_timeout = Some(v);
80 info!("use {} = {}", p[0], v);
81 }
82 } else if el.starts_with("--max_batch_size") {
83 let p: Vec<&str> = el.split('=').collect();
84 if let Ok(v) = p[1].parse::<u32>() {
85 max_batch_size = Some(v);
86 println!("use {} = {}", p[0], v);
87 }
88 } else if el.starts_with("--notify_channel_read_timeout") {
89 let p: Vec<&str> = el.split('=').collect();
90 if let Ok(v) = p[1].parse::<u64>() {
91 notify_channel_read_timeout = Some(v);
92 info!("use {} = {} ms", p[0], v);
93 }
94 } else if el.starts_with("--notify_channel_url") {
95 let p: Vec<&str> = el.split('=').collect();
96 notify_channel_url = p[1].to_owned();
97 }
98 }
99
100 if notify_channel_url.is_empty() {
101 if let Some(s) = Module::get_property("notify_channel_url") {
102 notify_channel_url = s;
103 }
104 }
105
106 let onto_types = vec![
107 "rdfs:Class",
108 "owl:Class",
109 "rdfs:Datatype",
110 "owl:Ontology",
111 "rdf:Property",
112 "owl:DatatypeProperty",
113 "owl:ObjectProperty",
114 "owl:OntologyProperty",
115 "owl:AnnotationProperty",
116 "v-ui:PropertySpecification",
117 "v-ui:DatatypePropertySpecification",
118 "v-ui:ObjectPropertySpecification",
119 "v-ui:TemplateSpecification",
120 "v-ui:ClassModel",
121 ]
122 .into_iter()
123 .map(|s| s.to_string())
124 .collect();
125
126 Module {
127 queue_prepared_count: 0,
128 notify_channel_url,
129 is_ready_notify_channel: false,
130 max_timeout_between_batches,
131 min_batch_size_to_cancel_timeout,
132 max_batch_size,
133 subsystem_id: module_id,
134 notify_channel_read_timeout,
135 syssig_ch: None,
136 name: module_name.to_owned(),
137 onto_types,
138 queue_part_retry_count: DEFAULT_QUEUE_PART_RETRY_COUNT,
139 queue_part_retry_pause_ms: DEFAULT_QUEUE_PART_RETRY_PAUSE_MS,
140 }
141 }
142
143 pub fn set_queue_part_retry(&mut self, count: u32, pause_ms: u64) {
144 self.queue_part_retry_count = count.max(1);
145 self.queue_part_retry_pause_ms = pause_ms;
146 }
147
148 pub fn new() -> Self {
149 Module::create(None, "")
150 }
151
152 pub fn is_content_onto(
153 &self,
154 cmd: IndvOp,
155 new_state: &mut Individual,
156 prev_state: &mut Individual,
157 ) -> bool {
158 if cmd != IndvOp::Remove {
159 if new_state.any_exists_v("rdf:type", &self.onto_types) {
160 return true;
161 }
162 } else if prev_state.any_exists_v("rdf:type", &self.onto_types) {
163 return true;
164 }
165 false
166 }
167
168 pub fn get_property<T: FromStr>(in_param: &str) -> Option<T> {
169 let conf = Ini::load_from_file("veda.properties").expect("fail load veda.properties file");
170
171 let aliases = conf
172 .section(Some("alias"))
173 .expect("fail parse veda.properties, section [alias]");
174
175 let args: Vec<String> = env::args().collect();
176
177 let params = [in_param.replace('_', "-"), in_param.replace('-', "_")];
178
179 for el in args.iter() {
180 for param in ¶ms {
181 if el.starts_with(&format!("--{}", param)) {
182 let p: Vec<&str> = el.split('=').collect();
183
184 if p.len() == 2 {
185 let v = p[1].trim();
186 let val = if let Some(a) = aliases.get(v) {
187 info!("use arg --{}={}, alias={}", param, a, v);
188 a
189 } else {
190 info!("use arg --{}={}", param, v);
191 v
192 };
193
194 return val.parse().ok();
195 }
196 }
197 }
198 }
199
200 let section = conf.section(None::<String>).expect("fail parse veda.properties");
201
202 if let Some(v) = section.get(in_param) {
203 let mut val = v.trim().to_owned();
204
205 if val.starts_with('$') {
206 if let Ok(val4var) = env::var(val.strip_prefix('$').unwrap_or_default()) {
207 info!("get env variable [{}]", val);
208 val = val4var;
209 } else {
210 info!("not found env variable {}", val);
211 return None;
212 }
213 }
214
215 let res = if let Some(a) = aliases.get(&val) {
216 info!("use param [{}]={}, alias={}", in_param, a, val);
217 a
218 } else {
219 info!("use param [{}]={}", in_param, val);
220 &val
221 };
222
223 return res.parse().ok();
224 }
225
226 error!("param [{}] not found", in_param);
227 None
228 }
229
230 pub(crate) fn get_info_of_part_with_retry(&self, queue_consumer: &mut Consumer) -> bool {
231 let max_attempts = self.queue_part_retry_count.max(1);
232 let pause = Duration::from_millis(self.queue_part_retry_pause_ms);
233
234 for attempt in 1..=max_attempts {
235 match queue_consumer
236 .queue
237 .get_info_of_part(queue_consumer.id, true)
238 {
239 Ok(()) => return true,
240 Err(e) => {
241 let err_str = e.as_str();
242 if attempt == 1 {
243 error!(
244 "{} get_info_of_part {}: {}",
245 self.queue_prepared_count, queue_consumer.id, err_str
246 );
247 } else if attempt < max_attempts {
248 warn!(
249 "get_info_of_part {}: {}, retry {}/{}",
250 queue_consumer.id, err_str, attempt, max_attempts
251 );
252 } else {
253 error!(
254 "get_info_of_part {}: {}, all {} attempts failed",
255 queue_consumer.id, err_str, max_attempts
256 );
257 }
258 if attempt < max_attempts {
259 thread::sleep(pause);
260 }
261 }
262 }
263 }
264
265 thread::sleep(pause);
266 false
267 }
268
269 pub fn connect_to_notify_channel(&mut self) -> Option<Socket> {
270 if !self.is_ready_notify_channel && !self.notify_channel_url.is_empty() {
271 let soc = Socket::new(Protocol::Sub0).unwrap();
272
273 let timeout = if let Some(t) = self.notify_channel_read_timeout {
274 t
275 } else {
276 1000
277 };
278
279 if let Err(e) = soc.set_opt::<RecvTimeout>(Some(Duration::from_millis(timeout))) {
280 error!("fail set timeout, {} err={}", self.notify_channel_url, e);
281 return None;
282 }
283
284 if let Err(e) = soc.dial(&self.notify_channel_url) {
285 error!("fail connect to, {} err={}", self.notify_channel_url, e);
286 return None;
287 } else {
288 let all_topics = vec![];
289 if let Err(e) = soc.set_opt::<Subscribe>(all_topics) {
290 error!("fail subscribe, {} err={}", self.notify_channel_url, e);
291 soc.close();
292 self.is_ready_notify_channel = false;
293 return None;
294 } else {
295 info!(
296 "success subscribe on queue changes: {}",
297 self.notify_channel_url
298 );
299 self.is_ready_notify_channel = true;
300 return Some(soc);
301 }
302 }
303 }
304 None
305 }
306
307 pub fn listen_queue_raw<T, B>(
308 &mut self,
309 queue_consumer: &mut Consumer,
310 module_context: &mut T,
311 before_batch: &mut fn(&mut B, &mut T, batch_size: u32) -> Option<u32>,
312 prepare: &mut fn(&mut B, &mut T, &RawObj, &Consumer) -> Result<bool, PrepareError>,
313 after_batch: &mut fn(&mut B, &mut T, prepared_batch_size: u32) -> Result<bool, PrepareError>,
314 heartbeat: &mut fn(&mut B, &mut T) -> Result<(), PrepareError>,
315 backend: &mut B,
316 ) {
317 self.listen_queue_comb(
318 queue_consumer,
319 module_context,
320 before_batch,
321 Some(prepare),
322 None,
323 after_batch,
324 heartbeat,
325 backend,
326 )
327 }
328
329 pub fn listen_queue<T, B>(
330 &mut self,
331 queue_consumer: &mut Consumer,
332 module_context: &mut T,
333 before_batch: &mut fn(&mut B, &mut T, batch_size: u32) -> Option<u32>,
334 prepare: &mut fn(&mut B, &mut T, &mut Individual, &Consumer) -> Result<bool, PrepareError>,
335 after_batch: &mut fn(&mut B, &mut T, prepared_batch_size: u32) -> Result<bool, PrepareError>,
336 heartbeat: &mut fn(&mut B, &mut T) -> Result<(), PrepareError>,
337 backend: &mut B,
338 ) {
339 self.listen_queue_comb(
340 queue_consumer,
341 module_context,
342 before_batch,
343 None,
344 Some(prepare),
345 after_batch,
346 heartbeat,
347 backend,
348 )
349 }
350
351 fn listen_queue_comb<T, B>(
352 &mut self,
353 queue_consumer: &mut Consumer,
354 module_context: &mut T,
355 before_batch: &mut fn(&mut B, &mut T, batch_size: u32) -> Option<u32>,
356 prepare_raw: Option<&mut fn(&mut B, &mut T, &RawObj, &Consumer) -> Result<bool, PrepareError>>,
357 prepare_indv: Option<
358 &mut fn(&mut B, &mut T, &mut Individual, &Consumer) -> Result<bool, PrepareError>,
359 >,
360 after_batch: &mut fn(&mut B, &mut T, prepared_batch_size: u32) -> Result<bool, PrepareError>,
361 heartbeat: &mut fn(&mut B, &mut T) -> Result<(), PrepareError>,
362 backend: &mut B,
363 ) {
364 if let Ok(ch) = sys_sig_listener() {
365 self.syssig_ch = Some(ch);
366 }
367
368 let mut soc = None;
369 let mut count_timeout_error = 0;
370
371 let mut prev_batch_time = Instant::now();
372 let update = tick(Duration::from_millis(1));
373 loop {
374 if let Some(qq) = &self.syssig_ch {
375 select! {
376 recv(update) -> _ => {
377 }
378 recv(qq) -> _ => {
379 info!("queue {}/{}, part:{}, pos:{}", queue_consumer.queue.base_path, queue_consumer.name, queue_consumer.id, queue_consumer.count_popped);
380 info!("Exit");
381 std::process::exit(exitcode::OK);
382 }
383 }
384 }
385
386 if let Err(PrepareError::Fatal) = heartbeat(backend, module_context) {
387 error!("heartbeat: found fatal error, stop listen queue");
388 break;
389 }
390
391 if soc.is_none() {
392 soc = self.connect_to_notify_channel();
393 if soc.is_none() {
394 thread::sleep(time::Duration::from_millis(NOTIFY_CHANNEL_RECONNECT_TIMEOUT));
395 info!("sleep {} ms", NOTIFY_CHANNEL_RECONNECT_TIMEOUT);
396 }
397 }
398
399 if !self.get_info_of_part_with_retry(queue_consumer) {
400 continue;
401 }
402
403 let size_batch = queue_consumer.get_batch_size();
404
405 let mut max_size_batch = size_batch;
406 if let Some(m) = self.max_batch_size {
407 max_size_batch = m;
408 }
409
410 if size_batch > 0 {
411 debug!("queue: batch size={}", size_batch);
412 if let Some(new_size) = before_batch(backend, module_context, size_batch) {
413 max_size_batch = new_size;
414 }
415 }
416
417 let mut prepared_batch_size = 0;
418 for _it in 0..max_size_batch {
419 if !queue_consumer.pop_header() {
420 break;
421 }
422
423 let mut raw = RawObj::new(vec![0; queue_consumer.record_len()]);
424
425 if let Err(e) = queue_consumer.pop_body(&mut raw.data) {
426 match e {
427 ErrorQueue::FailReadTailMessage => {
428 break;
429 }
430 ErrorQueue::InvalidChecksum => {
431 error!("[module] consumer:pop_body: invalid CRC, attempt seek next record");
432 queue_consumer.seek_next_pos();
433 break;
434 }
435 _ => {
436 error!(
437 "{} get msg from queue: {}",
438 self.queue_prepared_count,
439 e.as_str()
440 );
441 break;
442 }
443 }
444 }
445
446 let mut need_commit = true;
447
448 if let Some(&mut f) = prepare_raw {
449 match f(backend, module_context, &raw, queue_consumer) {
450 Err(e) => {
451 if let PrepareError::Fatal = e {
452 warn!("prepare: found fatal error, stop listen queue");
453 return;
454 }
455 }
456 Ok(b) => {
457 need_commit = b;
458 }
459 }
460 }
461
462 if let Some(&mut f) = prepare_indv {
463 let mut queue_element = Individual::new_raw(raw);
464 if parse_raw(&mut queue_element).is_ok() {
465 let mut is_processed = true;
466 if let Some(assigned_subsystems) =
467 queue_element.get_first_integer("assigned_subsystems")
468 {
469 if assigned_subsystems > 0 {
470 if let Some(my_subsystem_id) = self.subsystem_id {
471 if assigned_subsystems & my_subsystem_id == 0 {
472 is_processed = false;
473 }
474 } else {
475 is_processed = false;
476 }
477 }
478 }
479
480 if is_processed {
481 match f(backend, module_context, &mut queue_element, queue_consumer) {
482 Err(e) => {
483 if let PrepareError::Fatal = e {
484 warn!("prepare: found fatal error, stop listen queue");
485 return;
486 }
487 }
488 Ok(b) => {
489 need_commit = b;
490 }
491 }
492 }
493 }
494 }
495
496 if need_commit {
497 queue_consumer.commit();
498 }
499
500 self.queue_prepared_count += 1;
501
502 if self.queue_prepared_count % 1000 == 0 {
503 info!("get from queue, count: {}", self.queue_prepared_count);
504 }
505 prepared_batch_size += 1;
506 }
507
508 if size_batch > 0 {
509 match after_batch(backend, module_context, prepared_batch_size) {
510 Ok(b) => {
511 if b {
512 queue_consumer.commit();
513 }
514 }
515 Err(e) => {
516 if let PrepareError::Fatal = e {
517 warn!("after_batch: found fatal error, stop listen queue");
518 return;
519 }
520 }
521 }
522 }
523
524 if prepared_batch_size == size_batch {
525 if let Some(s) = &soc {
526 let wmsg = s.recv();
527 if let Err(e) = wmsg {
528 debug!("fail recv from queue notify channel, err={:?}", e);
529
530 if count_timeout_error > 0 && size_batch > 0 {
531 warn!("queue changed but we not received notify message, need reconnect...");
532 self.is_ready_notify_channel = false;
533 count_timeout_error += 1;
534 }
535 } else {
536 count_timeout_error = 0;
537 }
538 }
539 }
540
541 if let Some(t) = self.max_timeout_between_batches {
542 let delta = prev_batch_time.elapsed().as_millis() as u64;
543 if let Some(c) = self.min_batch_size_to_cancel_timeout {
544 if prepared_batch_size < c && delta < t {
545 thread::sleep(time::Duration::from_millis(t - delta));
546 info!("sleep {} ms", t - delta);
547 }
548 } else if delta < t {
549 thread::sleep(time::Duration::from_millis(t - delta));
550 info!("sleep {} ms", t - delta);
551 }
552 }
553
554 prev_batch_time = Instant::now();
555 }
556 }
557}
558
559pub fn get_inner_binobj_as_individual<'a>(
560 queue_element: &'a mut Individual,
561 field_name: &str,
562 new_indv: &'a mut Individual,
563) -> bool {
564 let binobj = queue_element.get_first_binobj(field_name);
565 if binobj.is_some() {
566 new_indv.set_raw(&binobj.unwrap_or_default());
567 if parse_raw(new_indv).is_ok() {
568 return true;
569 }
570 }
571 false
572}
573
574pub fn get_cmd(queue_element: &mut Individual) -> Option<IndvOp> {
575 let wcmd = queue_element.get_first_integer("cmd");
576 wcmd?;
577
578 Some(IndvOp::from_i64(wcmd.unwrap_or_default()))
579}
580
581pub fn init_log(module_name: &str) {
582 init_log_with_filter(module_name, None)
583}
584
585pub fn init_log_with_filter(module_name: &str, filter: Option<&str>) {
586 init_log_with_params(module_name, filter, false);
587}
588
589pub fn init_log_with_params(module_name: &str, filter: Option<&str>, with_thread_id: bool) {
590 let var_log_name = module_name.to_owned() + "_LOG";
591
592 let filters_str = match env::var(&var_log_name) {
593 Ok(val) if !val.is_empty() => {
594 println!("use env var: {}: {:?}", var_log_name, val);
595 val
596 }
597 _ => filter.unwrap_or("info").to_owned(),
598 };
599
600 if with_thread_id {
601 Builder::new()
602 .format(|buf, record| {
603 writeln!(
604 buf,
605 "{} {} [{}] - {}",
606 thread_id::get(),
607 Local::now().format("%Y-%m-%dT%H:%M:%S%.3f"),
608 record.level(),
609 record.args()
610 )
611 })
612 .parse_filters(&filters_str)
613 .try_init()
614 .unwrap_or(())
615 } else {
616 Builder::new()
617 .format(|buf, record| {
618 writeln!(
619 buf,
620 "{} [{}] - {}",
621 Local::now().format("%Y-%m-%dT%H:%M:%S%.3f"),
622 record.level(),
623 record.args()
624 )
625 })
626 .parse_filters(&filters_str)
627 .try_init()
628 .unwrap_or(())
629 }
630}
631
632pub fn get_info_of_module(module_name: &str) -> Option<(i64, i64)> {
633 let module_info = ModuleInfo::new("./data", module_name, false);
634 if module_info.is_err() {
635 error!(
636 "fail open info of [{}], err={:?}",
637 module_name,
638 module_info.err()
639 );
640 return None;
641 }
642
643 let mut info = module_info.unwrap();
644 info.read_info()
645}