eis/eqpalg/utility/alarm_poster.cc

#include "mix_cc/json.h"
#include "mix_cc/sql.h"
#include "mix_cc/sql/database/db2_t.h"
#include "mix_cc/type/mix_time.h"
#include <eqpalg/table_struct/t_rule_record_time.h>
#include <eqpalg/table_struct/t_rule_result.h>
#include <eqpalg/utility/alarm_poster.h>
#include <eqpalg/utility/build_json_from_alarm_Info.h>
#include <mutex>
#include <zlib/MemQueue.hpp>
#include <zlib/MemVar.h>
namespace utility {
/**
 * @brief 报警信息队列
 * 只写
 */
namespace {
// std::mutex local_mutext{};      ///<共享锁
const int AlarmLen = 4000;      ///<报警信息的最大长度
const int AlarmMaxSize = 10000; ///<报警信息队列容量
typedef char AlarmJsonData[AlarmLen];
AlarmJsonData alarm_json_data_char;
CMemQueue<AlarmJsonData> alarm_json_data_queue =
    CMemQueue<AlarmJsonData>("ZONE0", AlarmMaxSize);
} // namespace

AlarmPoster::AlarmPoster()
    : gb_logger_(std::make_unique<GbLogger>("AlarmPoster")) {
  gb_logger_->log_info("AlarmPoster()--test1");
  message_queue_proxy_ = ProxyMag::GetAppICEPrx("baosight/zmqp");
  gb_logger_->log_info("AlarmPoster()--test2");
}

AlarmPoster::~AlarmPoster() {}

int AlarmPoster::alarm(
    const AlarmInfo &alarm_info,
    std::chrono::system_clock::time_point *const prev_alarm_time) {
  // 2021-10-27 报警间隔时间
  int minutesTh = std::max(5, CMemVar::Const()->AlarmIntervalHours * 60);
  if (alarm_info.alarm_end_time - *prev_alarm_time > minutes(minutesTh)) {
    *prev_alarm_time = alarm_info.alarm_end_time;
    // std::lock_guard<std::mutex> guard(local_mutext);
    auto alarm_json1 = build_json_from_alarm_Info(alarm_info);
    strcpy(alarm_json_data_char, alarm_json1.c_str());
    alarm_json_data_queue.push(&alarm_json_data_char);
    gb_logger_->log_info("alarm_json1:"+alarm_json1);
  }

  //   //从 t_rule_record_time 中查询 模型修改时间  模型启动时间
  //   mix_cc::mix_time_t rule_lmt;
  //   mix_cc::mix_time_t rule_lst;
  //   T_RULE_RECORD_TIME t_rule_record_time;
  //   auto select_statement =
  //       select(t_rule_record_time.tos(), t_rule_record_time.rule_tom())
  //           .from(t_rule_record_time)
  //           .where(t_rule_record_time.ruleId() = alarm_info.cfg_info.id);

  //   auto tos_tom_maybe =
  //       mix_cc::sql::exec<db2_t, T_RULE_RECORD_TIME>(select_statement);
  //   if (tos_tom_maybe.is_just()) {
  //     auto tos_tom = tos_tom_maybe.unsafe_get_just();
  //     if (tos_tom.size() == 0) {
  //       gb_logger_->log_error("t_rule_record_time表格无记录");
  //       rule_lmt = mix_cc::mix_time_t(std::chrono::system_clock::now());
  //       rule_lst = mix_cc::mix_time_t(std::chrono::system_clock::now());
  //     } else {
  //       rule_lmt = tos_tom[0].rule_tom;
  //       rule_lst = tos_tom[0].tos;
  //     }

  //   } else {
  //     gb_logger_->log_error("t_rule_record_time表格查询失败");
  //   }

  //   T_RULE_RESULT rr;
  //   // 向表中插入
  //   auto num = exec<db2_t, size_t>(insert_into(rr).set(
  //       rr.ruleId() = alarm_info.cfg_info.id,
  //       rr.ruleName() = alarm_info.cfg_info.name,
  //       rr.ruleGroup() = alarm_info.cfg_info.group,
  //       rr.ruleBtime() = mix_cc::mix_time_t(alarm_info.alarm_start_time),
  //       rr.ruleEtime() = mix_cc::mix_time_t(alarm_info.alarm_end_time),
  //       rr.dealResult() = 0,
  //       rr.result() = mix_cc::to_string(alarm_info.content),
  //       rr.ruleLastMTime() = rule_lmt, rr.ruleLastSTime() = rule_lst));
  //   auto alarm_json = build_json_from_alarm_Info(alarm_info);
  //   gb_logger_->log_info(alarm_json.c_str());
  //   if (num.is_nothing()) {
  //     this->gb_logger_->log_error("数据库存储异常");
  //   }
  //   try {
  //     std::vector<unsigned char> seq(
  //         (unsigned char*)alarm_json.c_str(),
  //         (unsigned char*)alarm_json.c_str() + alarm_json.length());
  //     message_queue_proxy_->SendDataShort(911, seq, alarm_json.length());
  //   } catch (::Ice::LocalException& e) {
  //     gb_logger_->log_error(std::string("消息队列发送失败,请检查消息") +
  //                           e.what());
  //   }
  // }
  return 0;
}

int AlarmPoster::alarm(const std::string alarm_info) {
  try {
    strcpy(alarm_json_data_char, alarm_info.c_str());
    alarm_json_data_queue.push(&alarm_json_data_char);
    return 0;
  } catch (std::exception &e) {
    gb_logger_->log_error(e.what());
    return -1;
  }
  return -1;
}

int AlarmPoster::alarm(const AlarmInfo &alarm_info, string processType) {
  try {
    if (processType == "task") {
      auto alarm_json = build_json_from_alarm_Info(alarm_info);
      std::vector<unsigned char> seq((unsigned char *)alarm_json.c_str(),
                                     (unsigned char *)alarm_json.c_str() +
                                         alarm_json.length());
      message_queue_proxy_->SendDataShort(911, seq, alarm_json.length());
    }

  } catch (::Ice::LocalException &e) {
    gb_logger_->log_error("消息队列发送失败,请检查消息");
  }
  return 0;
}

int AlarmPoster::alarm(const AlarmInfo &alarm_info,
                       std::chrono::system_clock::time_point *prev_alarm_time,
                       string remark) {
  if (alarm_info.alarm_end_time - *prev_alarm_time > minutes(5)) {
    *prev_alarm_time = alarm_info.alarm_end_time;
    //从 t_rule_record_time 中查询 模型修改时间  模型启动时间
    mix_cc::mix_time_t rule_lmt;
    mix_cc::mix_time_t rule_lst;
    T_RULE_RECORD_TIME t_rule_record_time;
    auto select_statement =
        select(t_rule_record_time.tos(), t_rule_record_time.rule_tom())
            .from(t_rule_record_time)
            .where(t_rule_record_time.ruleId() = alarm_info.cfg_info.id);

    auto tos_tom_maybe =
        mix_cc::sql::exec<db2_t, T_RULE_RECORD_TIME>(select_statement);
    if (tos_tom_maybe.is_just()) {
      auto tos_tom = tos_tom_maybe.unsafe_get_just();
      rule_lmt = tos_tom[0].rule_tom;
      rule_lst = tos_tom[0].tos;
    } else {
      gb_logger_->log_info("t_rule_record_time表格查询失败");
    }

    T_RULE_RESULT rr;
    // 向表中插入
    auto num = exec<db2_t, size_t>(insert_into(rr).set(
        rr.ruleId() = alarm_info.cfg_info.id,
        rr.ruleName() = alarm_info.cfg_info.name,
        rr.ruleGroup() = alarm_info.cfg_info.group,
        rr.ruleBtime() = mix_cc::mix_time_t(alarm_info.alarm_start_time),
        rr.ruleEtime() = mix_cc::mix_time_t(alarm_info.alarm_end_time),
        rr.dealResult() = 0,
        rr.result() = mix_cc::to_string(alarm_info.content),
        rr.ruleLastMTime() = rule_lmt, rr.ruleLastSTime() = rule_lst));
    auto alarm_json = build_json_from_alarm_Info(alarm_info, remark);
    gb_logger_->log_info(alarm_json.c_str());
    if (num.is_nothing()) {
      this->gb_logger_->log_error("数据库存储异常");
    }
    try {
      std::vector<unsigned char> seq((unsigned char *)alarm_json.c_str(),
                                     (unsigned char *)alarm_json.c_str() +
                                         alarm_json.length());
      message_queue_proxy_->SendDataShort(911, seq, alarm_json.length());
    } catch (::Ice::LocalException &e) {
      gb_logger_->log_error("消息队列发送失败,请检查消息");
    }
  }
  return 0;
}

bool AlarmPoster::zmqp_send(int event_no, const string &content) {
  try {
    std::vector<unsigned char> seq((unsigned char *)content.c_str(),
                                   (unsigned char *)content.c_str() +
                                       content.length());
    message_queue_proxy_->SendDataShort(event_no, seq, content.length());
    return true;
  } catch (::Ice::LocalException &e) {
    gb_logger_->log_error("消息队列发送失败,请检查消息");
    return false;
  }
}

} // namespace utility