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 {
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);
}

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) {
  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;
    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);
  }
  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;
    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;
  }
}

}