钟来 / Luhui · 提交

转到一个项目

作者 钟来
提交 27f5de25c0c5cf8bd8e6c317a9208e89ea7a25cb 1 个父辈 afaf61fa

开放新版的告警系统

内嵌 并排对比
正在显示 66 个修改的文件 包含 2196 行增加 和 286 行删除
package com.ruoyi.system.domain.user;
import io.swagger.annotations.ApiModel;
import io.swagger.annotations.ApiModelProperty;
@ApiModel("终端分组")
public class UserOpenid {
@ApiModelProperty("主键")
private Integer id; // int DEFAULT NULL COMMENT '主键',
@ApiModelProperty("用户id")
private Integer user_id; // int DEFAULT NULL COMMENT '用户id',
@ApiModelProperty("微信用户的唯一标识")
private String openid; // varchar(50) DEFAULT NULL COMMENT '微信用户的唯一标识',
@ApiModelProperty("微信平台的配置参数表id")
private Integer xcx_config_id; // int DEFAULT NULL COMMENT '微信平台的配置参数表id',
@ApiModelProperty("创建时间")
private String create_time ; //timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP COMMENT '创建时间',
public String getCreate_time() {
return create_time;
}
public void setCreate_time(String create_time) {
this.create_time = create_time;
}
public Integer getId() {
return id;
}
public void setId(Integer id) {
this.id = id;
}
public Integer getUser_id() {
return user_id;
}
public void setUser_id(Integer user_id) {
this.user_id = user_id;
}
public String getOpenid() {
return openid;
}
public void setOpenid(String openid) {
this.openid = openid;
}
public Integer getXcx_config_id() {
return xcx_config_id;
}
public void setXcx_config_id(Integer xcx_config_id) {
this.xcx_config_id = xcx_config_id;
}
}
... ...
... ... @@ -71,7 +71,7 @@ public class SecurityConfigService {
// 过滤请求
.authorizeRequests()
// 对于登录login 注册register 验证码captchaImage 允许匿名访问
.antMatchers(antMatchers.split(",")).anonymous()
.antMatchers(antMatchers.split(",")).permitAll()
.antMatchers(
HttpMethod.GET,
"/",
... ... @@ -81,11 +81,7 @@ public class SecurityConfigService {
"/**/*.js",
"/profile/**"
).permitAll()
.antMatchers("/swagger-ui.html").anonymous()
.antMatchers("/swagger-resources/**").anonymous()
.antMatchers("/webjars/**").anonymous()
.antMatchers("/*/api-docs").anonymous()
.antMatchers("/druid/**").anonymous()
.antMatchers("/swagger-ui.html","/swagger-resources/**","/webjars/**","/*/api-docs","/druid/**").permitAll()
// 除上面外的所有请求全部需要鉴权认证
.anyRequest().authenticated()
.and()
... ...
... ... @@ -18,7 +18,7 @@ public class Topic {
private String topicType;
private String messageid;
private String payloadtype;
private boolean isAllup;
public Topic() {
}
... ...
... ... @@ -46,6 +46,16 @@ public class IotAlertLog extends BaseEntity
private Long create_time; // datetime DEFAULT NULL COMMENT '创建时间',
@ApiModelProperty("类型（1=告警，2=场景联动）")
private Integer type; // tinyint DEFAULT NULL COMMENT '类型（1=告警，2=场景联动）',
@ApiModelProperty("关联的用户id（0为系统生成）")
private Integer user_id;
public Integer getUser_id() {
return user_id;
}
public void setUser_id(Integer user_id) {
this.user_id = user_id;
}
public Long getAlertId() {
return alertId;
... ...
package com.zhonglai.luhui.device.domain;
import io.swagger.annotations.ApiModel;
import io.swagger.annotations.ApiModelProperty;
/**
* 告警通知渠道
*/
@ApiModel("告警通知渠道")
public class IotAlertNoticeChannel {
@ApiModelProperty("主键")
private Integer id; // int NOT NULL AUTO_INCREMENT COMMENT '主键',
@ApiModelProperty("渠道类型(1公众号，2小程序，3鱼儿乐app，4短信，5电话)")
private Integer type; // int NOT NULL COMMENT '渠道类型(1公众号，2小程序，3鱼儿乐app，4短信，5电话)',
@ApiModelProperty("渠道配置")
private String config; // json NOT NULL COMMENT '渠道配置',
@ApiModelProperty("名称")
private String name; // varchar(50) NOT NULL COMMENT '名称',
@ApiModelProperty("创建时间")
private String create_time; // varchar(50) NOT NULL COMMENT '创建时间',
@ApiModelProperty("修改时间")
private String update_time; // varchar(50) NOT NULL COMMENT '修改时间',
public String getCreate_time() {
return create_time;
}
public void setCreate_time(String create_time) {
this.create_time = create_time;
}
public String getUpdate_time() {
return update_time;
}
public void setUpdate_time(String update_time) {
this.update_time = update_time;
}
public Integer getId() {
return id;
}
public void setId(Integer id) {
this.id = id;
}
public Integer getType() {
return type;
}
public void setType(Integer type) {
this.type = type;
}
public String getConfig() {
return config;
}
public void setConfig(String config) {
this.config = config;
}
public String getName() {
return name;
}
public void setName(String name) {
this.name = name;
}
}
... ...
package com.zhonglai.luhui.device.domain;
import io.swagger.annotations.ApiModel;
import io.swagger.annotations.ApiModelProperty;
/**
* 用户告警通知配置
*/
@ApiModel("用户告警通知配置")
public class IotAlertUserNotice {
@ApiModelProperty("主键")
private Integer id; // int NOT NULL AUTO_INCREMENT COMMENT '主键id',
@ApiModelProperty("通知渠道(iot_alert_notice_channel表的id集合,多个英文逗号分割)")
private String channels; // varchar(100) DEFAULT NULL COMMENT '通知渠道(iot_alert_notice_channel表的id集合,多个英文逗号分割)',
@ApiModelProperty("告警类型（1系统告警，2用户告警）")
private Integer type; // int DEFAULT NULL COMMENT '告警类型（1系统告警，2用户告警）',
@ApiModelProperty("告警id集合（关联iot_alert_log和iot_alert_user表的id,多个英文逗号分割，*表示所有）")
private String alert_ids; // varchar(100) DEFAULT '*' COMMENT '告警id集合（关联iot_alert_log和iot_alert_user表的id,多个英文逗号分割，*表示所有）'
@ApiModelProperty("用户id")
private Integer user_id; // int DEFAULT NULL COMMENT '用户id'
@ApiModelProperty("创建时间")
private String create_time; // int DEFAULT NULL COMMENT '创建时间'
@ApiModelProperty("更新时间")
private String update_time; // int DEFAULT NULL COMMENT '更新时间'
public Integer getId() {
return id;
}
public void setId(Integer id) {
this.id = id;
}
public Integer getType() {
return type;
}
public void setType(Integer type) {
this.type = type;
}
public Integer getUser_id() {
return user_id;
}
public void setUser_id(Integer user_id) {
this.user_id = user_id;
}
public String getCreate_time() {
return create_time;
}
public void setCreate_time(String create_time) {
this.create_time = create_time;
}
public String getUpdate_time() {
return update_time;
}
public void setUpdate_time(String update_time) {
this.update_time = update_time;
}
public String getChannels() {
return channels;
}
public void setChannels(String channels) {
this.channels = channels;
}
public String getAlert_ids() {
return alert_ids;
}
public void setAlert_ids(String alert_ids) {
this.alert_ids = alert_ids;
}
}
... ...
... ... @@ -127,6 +127,16 @@ public class IotDevice implements Serializable
@ApiModelProperty("数据更新时间")
private Integer data_update_time;
@ApiModelProperty("操作令牌")
private String operation_token;
public String getOperation_token() {
return operation_token;
}
public void setOperation_token(String operation_token) {
this.operation_token = operation_token;
}
public Integer getData_update_time() {
return data_update_time;
... ...
... ... @@ -64,6 +64,16 @@ public class IotTerminal implements Serializable
@ApiModelProperty("在线状态")
private Integer online;
@ApiModelProperty("操作令牌")
private String operation_token;
public String getOperation_token() {
return operation_token;
}
public void setOperation_token(String operation_token) {
this.operation_token = operation_token;
}
private UserTerminalGroupRelation userTerminalGroupRelation;
private IotDevice iotDevice;
... ...
package com.zhonglai.luhui.device.dto;
/**
* 公众号通知的配置
*/
public class WeiXinGongZhongHaoNoticeDto {
private Integer xcxConfigId;
private String xcxappid;
private String xcxsecret;
private String gzhappid;
private String gzhsecret;
private String templateId;
private String redirectUrl;
public String getGzhsecret() {
return gzhsecret;
}
public void setGzhsecret(String gzhsecret) {
this.gzhsecret = gzhsecret;
}
public Integer getXcxConfigId() {
return xcxConfigId;
}
public void setXcxConfigId(Integer xcxConfigId) {
this.xcxConfigId = xcxConfigId;
}
public String getXcxappid() {
return xcxappid;
}
public void setXcxappid(String xcxappid) {
this.xcxappid = xcxappid;
}
public String getXcxsecret() {
return xcxsecret;
}
public void setXcxsecret(String xcxsecret) {
this.xcxsecret = xcxsecret;
}
public String getGzhappid() {
return gzhappid;
}
public void setGzhappid(String gzhappid) {
this.gzhappid = gzhappid;
}
public String getTemplateId() {
return templateId;
}
public void setTemplateId(String templateId) {
this.templateId = templateId;
}
public String getRedirectUrl() {
return redirectUrl;
}
public void setRedirectUrl(String redirectUrl) {
this.redirectUrl = redirectUrl;
}
}
... ...
... ... @@ -67,5 +67,5 @@ public interface IIotTerminalService
List<IotTerminal> selectNotUserIotTerminalListByImei(String device_id);
int batchUpName(List<String> ids,List<String> names);
int batchUpName(List<String> ids,List<String> names,Integer user_id);
}
... ...
... ... @@ -155,8 +155,8 @@ public class IotTerminalServiceImpl implements IIotTerminalService
}
@Override
public int batchUpName(List<String> ids, List<String> names) {
StringBuffer sql = new StringBuffer("UPDATE `mqtt_broker`.`iot_terminal` SET `name` = CASE");
public int batchUpName(List<String> ids, List<String> names,Integer user_id) {
StringBuffer sql = new StringBuffer("UPDATE `mqtt_broker`.`iot_terminal` SET `user_info_id`="+user_id+",`name` = CASE");
StringBuffer where = new StringBuffer(" where id in(");
for (int i=0;i<ids.size();i++)
{
... ...
... ... @@ -5,7 +5,7 @@ PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN"
<mapper namespace="com.zhonglai.luhui.device.mapper.IotAlertLogMapper">
<resultMap type="com.zhonglai.luhui.device.domain.IotAlertLog" id="AlertLogResult">
<result property="alertLogId" column="alert_log__id" />
<result property="alertLogId" column="alert_log_id" />
<result property="alertName" column="alert_name" />
<result property="alertId" column="alert_id" />
<result property="alertLevel" column="alert_level" />
... ... @@ -13,10 +13,11 @@ PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN"
<result property="deviceId" column="device_id" />
<result property="create_time" column="create_time" />
<result property="type" column="type" />
<result property="user_id" column="user_id" />
</resultMap>
<sql id="selectAlertLogVo">
select alert_log__id, alert_name,alert_id, alert_level, status, device_id, create_time, type from iot_alert_log
select alert_log_id, alert_name,alert_id, alert_level, `status`, device_id, create_time, `type`,user_id from iot_alert_log
</sql>
<select id="selectAlertLogList" parameterType="com.zhonglai.luhui.device.domain.IotAlertLog" resultMap="AlertLogResult">
... ...
... ... @@ -34,11 +34,11 @@ PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN"
<result property="listen_service_ip" column="listen_service_ip" />
<result property="device_life" column="device_life" />
<result property="data_update_time" column="data_update_time" />
<result property="operation_token" column="operation_token" />
</resultMap>
<sql id="selectIotDeviceVo">
select active_time, client_id, completion_auth, create_by, create_time, del_flag, firmware_version, img_url, is_shadow, latitude, location_way, longitude, `name`, network_address, network_ip, remark, rssi, status, summary, things_model_value, update_by, update_time, product_id,mqtt_username,payload_type,things_model_config,listen_service_ip,device_life,data_update_time,mqtt_username from iot_device
select active_time, client_id, completion_auth, create_by, create_time, del_flag, firmware_version, img_url, is_shadow, latitude, location_way, longitude, `name`, network_address, network_ip, remark, rssi, status, summary, things_model_value, update_by, update_time, product_id,mqtt_username,payload_type,things_model_config,listen_service_ip,device_life,data_update_time,mqtt_username,operation_token from iot_device
</sql>
<select id="selectIotDeviceList" parameterType="IotDevice" resultMap="IotDeviceResult">
... ...
... ... @@ -5,11 +5,15 @@ import com.alibaba.fastjson.JSONObject;
import com.ruoyi.common.core.domain.DeviceCommandApi;
import com.ruoyi.common.core.domain.Message;
import com.ruoyi.common.core.domain.MessageCode;
import com.ruoyi.common.utils.StringUtils;
import lombok.val;
import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.exception.RequestTimeoutException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.remoting.common.RemotingHelper;
import org.apache.rocketmq.remoting.exception.RemotingException;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.slf4j.Logger;
... ... @@ -19,6 +23,8 @@ import org.springframework.beans.factory.annotation.Value;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Service;
import java.io.UnsupportedEncodingException;
@Service
public class RocketMqService {
... ... @@ -42,7 +48,11 @@ public class RocketMqService {
public Message send(String topic,byte[] payload)
{
org.apache.rocketmq.common.message.Message msg = new org.apache.rocketmq.common.message.Message(topic,sendTags,payload);
return send(topic,payload,null);
}
public Message send(String topic,byte[] payload,String operation_token)
{
org.apache.rocketmq.common.message.Message msg = new org.apache.rocketmq.common.message.Message(topic,null==operation_token?sendTags:operation_token,payload);
try {
org.apache.rocketmq.common.message.Message ms = rocketMQTemplate.getProducer().request(msg,30000l);
JSONObject jsonObject = (JSONObject) JSON.parse(new String(ms.getBody()));
... ... @@ -62,8 +72,9 @@ public class RocketMqService {
return new Message(MessageCode.DEFAULT_FAIL_CODE);
}
public void syncSend(String topic, Object message)
public void syncSend(String topic,byte[] payload,String operation_token)
{
rocketMQTemplate.syncSend(topic, MessageBuilder.withPayload(message).build(), 3000, 1);
rocketMQTemplate.sendOneWay(topic+":"+(null==operation_token?sendTags:operation_token),payload);
}
}
... ...
... ... @@ -8,8 +8,10 @@ import com.ruoyi.common.core.domain.Message;
import com.ruoyi.common.enums.BusinessType;
import com.ruoyi.common.utils.GsonConstructor;
import com.zhonglai.luhui.action.BaseController;
import com.zhonglai.luhui.device.domain.IotDevice;
import com.zhonglai.luhui.device.dto.CommandType;
import com.zhonglai.luhui.device.dto.DeviceCommand;
import com.zhonglai.luhui.device.service.IIotDeviceService;
import com.zhonglai.luhui.rocketmq.service.RocketMqService;
import io.swagger.annotations.Api;
import io.swagger.annotations.ApiImplicitParam;
... ... @@ -32,6 +34,9 @@ import java.io.IOException;
public class ControlDeviceConreoller extends BaseController {
@Autowired
private RocketMqService rocketMqService;
@Autowired
private IIotDeviceService iotDeviceService;
@ApiOperation(value = "写指令",notes = "body参数描述：\r\n" +
"{\n" +
" \"0\":{\n" +
... ... @@ -66,7 +71,46 @@ public class ControlDeviceConreoller extends BaseController {
deviceCommand.setDeviceId(deviceId);
deviceCommand.setCommandType(CommandType.write);
deviceCommand.setData(GsonConstructor.get().fromJson(body,JsonObject.class));
return control(deviceCommand);
return deviceControl(deviceCommand);
}
@ApiOperation(value = "读指令",notes = "body参数描述：\r\n" +
"{\n" +
" \"0\":{\n" +
" \"rssi\": -43,\n" +
" \"firmware_version\": 1.2,\n" +
" \"longitude\": 0,\n" +
" \"latitude\": 0,\n" +
" \"summary\": {\n" +
" \t \"name\": \"wumei-smart\",\n" +
" \t \"chip\": \"esp8266\",\n" +
" \t \"author\": \"kerwincui\",\n" +
" \t \"version\": 1.2,\n" +
" \t \"createTime\": \"2022-06-06\"\n" +
" }\n" +
" },\n" +
" \"1\":{\n" +
" \"id1\":\"value1\",\n" +
" \"id2\":\"value2\",\n" +
" \"id3\":\"value3\"\n" +
" }\n" +
"}")
@ApiImplicitParams({
@ApiImplicitParam(value = "网关id",name = "deviceId"),
})
@PreAuthorize("@ss.hasPermi('iot:controlDevice:write')")
@Log(title = "读指令", businessType = BusinessType.WRITE)
@PostMapping("/read/{deviceId}")
public AjaxResult read(@PathVariable String deviceId, HttpServletRequest request) throws IOException {
byte[] bodyBytes = StreamUtils.copyToByteArray(request.getInputStream());
String body = new String(bodyBytes, request.getCharacterEncoding());
DeviceCommand deviceCommand = new DeviceCommand();
deviceCommand.setDeviceId(deviceId);
deviceCommand.setCommandType(CommandType.write);
deviceCommand.setData(GsonConstructor.get().fromJson(body,JsonObject.class));
return deviceControl(deviceCommand);
}
@ApiOperation("更新缓存模型")
... ... @@ -80,7 +124,7 @@ public class ControlDeviceConreoller extends BaseController {
JsonObject jsonObject = new JsonObject();
jsonObject.addProperty("product_id",product_id);
deviceCommand.setData(jsonObject);
return control(deviceCommand);
return sysControl(deviceCommand);
}
@ApiOperation("更新缓存翻译模型")
... ... @@ -94,7 +138,7 @@ public class ControlDeviceConreoller extends BaseController {
JsonObject jsonObject = new JsonObject();
jsonObject.addProperty("product_id",product_id);
deviceCommand.setData(jsonObject);
return control(deviceCommand);
return sysControl(deviceCommand);
}
@ApiOperation("清除网关缓存")
... ... @@ -106,7 +150,7 @@ public class ControlDeviceConreoller extends BaseController {
DeviceCommand deviceCommand = new DeviceCommand();
deviceCommand.setDeviceId(deviceId);
deviceCommand.setCommandType(CommandType.cleanDeviceHost);
return control(deviceCommand);
return sysControl(deviceCommand);
}
@ApiOperation("更新终端缓存")
... ... @@ -124,7 +168,7 @@ public class ControlDeviceConreoller extends BaseController {
JsonObject jsonObject = new JsonObject();
jsonObject.addProperty("sensor_number",sensor_number);
deviceCommand.setData(jsonObject);
return control(deviceCommand);
return sysControl(deviceCommand);
}
@ApiOperation("添加订阅")
... ... @@ -142,7 +186,7 @@ public class ControlDeviceConreoller extends BaseController {
jsonObject.addProperty("product_ids",product_ids);
jsonObject.addProperty("ip",ip);
deviceCommand.setData(jsonObject);
return control(deviceCommand);
return sysControl(deviceCommand);
}
@ApiOperation(value = "通知",notes = "body参数描述：\r\n" +
... ... @@ -166,12 +210,28 @@ public class ControlDeviceConreoller extends BaseController {
deviceCommand.setDeviceId(deviceId);
deviceCommand.setCommandType(CommandType.notice);
deviceCommand.setData(GsonConstructor.get().fromJson(body,JsonObject.class));
return control(deviceCommand);
return deviceControl(deviceCommand);
}
private AjaxResult deviceControl(DeviceCommand deviceCommand)
{
IotDevice iotDevice = iotDeviceService.selectIotDeviceByClient_id(deviceCommand.getDeviceId());
if(null == iotDevice || iotDevice.getStatus()!=3)
{
return AjaxResult.error("设备不存在或者不在线");
}
Message message = rocketMqService.send("deviceCommandListen",GsonConstructor.get().toJson(deviceCommand).getBytes(),iotDevice.getOperation_token());
if(message.getCode()==1)
{
return new AjaxResult(HttpStatus.SUCCESS, message.getMessage(), message.getData());
}
return new AjaxResult(HttpStatus.ERROR, message.getMessage(), message.getData());
}
private AjaxResult control( DeviceCommand deviceCommand)
private AjaxResult sysControl(DeviceCommand deviceCommand)
{
Message message = rocketMqService.send("deviceCommandListen",GsonConstructor.get().toJson(deviceCommand).getBytes());
Message message = rocketMqService.send("deviceCommandListen",GsonConstructor.get().toJson(deviceCommand).getBytes(),"SysCommand");
if(message.getCode()==1)
{
return new AjaxResult(HttpStatus.SUCCESS, message.getMessage(), message.getData());
... ...
... ... @@ -2,10 +2,13 @@ package com.zhonglai.luhui.admin.controller.iot;
import cn.hutool.core.bean.BeanUtil;
import com.alibaba.fastjson.JSONObject;
import com.google.gson.JsonArray;
import com.google.gson.JsonElement;
import com.ruoyi.common.annotation.Log;
import com.ruoyi.common.core.domain.AjaxResult;
import com.ruoyi.common.core.page.TableDataInfo;
import com.ruoyi.common.enums.BusinessType;
import com.ruoyi.common.utils.GsonConstructor;
import com.ruoyi.common.utils.StringUtils;
import com.zhonglai.luhui.admin.dto.*;
import com.zhonglai.luhui.action.BaseController;
... ... @@ -87,7 +90,7 @@ public class IotAlertController extends BaseController
{
if (checkTriggers(alert))
{
return AjaxResult.error("触发器参数必填");
return AjaxResult.error("触发器参数校验失败");
}
// if (checkActions(alert))
// {
... ... @@ -107,7 +110,7 @@ public class IotAlertController extends BaseController
{
if (checkTriggers(alert))
{
return AjaxResult.error("触发器参数必填");
return AjaxResult.error("触发器参数校验失败");
}
// if (checkActions(alert))
// {
... ... @@ -132,7 +135,7 @@ public class IotAlertController extends BaseController
* 检验触发器
* @return
*/
private boolean checkTriggers(IotAlert alert)
private static boolean checkTriggers(IotAlert alert)
{
String triggers = Optional.ofNullable(alert).orElse(new IotAlert()).getTriggers();
if(StringUtils.isNotEmpty(triggers))
... ... @@ -143,8 +146,18 @@ public class IotAlertController extends BaseController
AttributeTriggers attributeTriggers = JSONObject.parseObject(triggers, AttributeTriggers.class);
return BeanUtil.hasNullField(attributeTriggers);
case 2:
TriggerTriggers triggerTriggers = JSONObject.parseObject(triggers, TriggerTriggers.class);
return BeanUtil.hasNullField(triggerTriggers);
JsonElement jsonElement = GsonConstructor.get().fromJson(triggers, JsonElement.class);
if(!jsonElement.isJsonArray())
{
return false;
}
JsonArray jsonArray = jsonElement.getAsJsonArray();
for (int i=0;i<jsonArray.size();i++)
{
TriggerTriggers triggerTriggers = JSONObject.parseObject(jsonArray.get(i).toString(), TriggerTriggers.class);
return BeanUtil.hasNullField(triggerTriggers);
}
case 3:
TimerTriggers timerTriggers = JSONObject.parseObject(triggers, TimerTriggers.class);
return BeanUtil.hasNullField(timerTriggers,"terminalids");
... ... @@ -154,18 +167,6 @@ public class IotAlertController extends BaseController
return false;
}
public static void main(String[] args) {
String triggers = "{\"model_name\":\"alarm\",\"valueMapName\":{{\"1\":\"传感器检测失败故障\"},{\"3\",\"超低氧告警\"}}}";
AttributeTriggers attributeTriggers = new AttributeTriggers();
attributeTriggers.setModel_name("alarm");
Map<Object,String> valueMapName = new HashMap<>();
valueMapName.put("1","传感器检测失败故障");
valueMapName.put("3","超低氧告警");
attributeTriggers.setValueMapName(valueMapName);
System.out.println(triggers);
System.out.println(JSONObject.toJSON(attributeTriggers));
}
/**
* 检验执行动作
* @return
... ...
package com.zhonglai.luhui.admin.controller.iot;
import cn.hutool.core.bean.BeanUtil;
import com.alibaba.fastjson.JSONObject;
import com.ruoyi.common.annotation.Log;
import com.ruoyi.common.core.domain.AjaxResult;
import com.ruoyi.common.core.page.TableDataInfo;
import com.ruoyi.common.enums.BusinessType;
import com.ruoyi.common.utils.StringUtils;
import com.zhonglai.luhui.action.BaseController;
import com.zhonglai.luhui.admin.dto.AttributeTriggers;
import com.zhonglai.luhui.dao.service.PublicService;
import com.zhonglai.luhui.device.domain.IotAlertLog;
import com.zhonglai.luhui.device.domain.IotAlertNoticeChannel;
import com.zhonglai.luhui.device.domain.IotPermission;
import com.zhonglai.luhui.device.dto.WeiXinGongZhongHaoNoticeDto;
import io.swagger.annotations.Api;
import io.swagger.annotations.ApiOperation;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.security.access.prepost.PreAuthorize;
import org.springframework.web.bind.annotation.*;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
@Api(tags = "告警通知渠道")
@RestController
@RequestMapping("/iot/alertNoticeChannel")
public class IotAlertNoticeChannelController extends BaseController {
@Autowired
private PublicService publicService;
/**
* 查询设备告警列表
*/
@ApiOperation("查询告警通知渠道列表")
@PreAuthorize("@ss.hasPermi('iot:alertNoticeChannel:list')")
@GetMapping("/list")
public TableDataInfo list(IotAlertNoticeChannel iotAlertNoticeChannel)
{
startPage();
Map<String, String> whereMap = new HashMap<>();
whereMap.put("name","like");
List<Map<String,Object>> list = publicService.getObjectList(iotAlertNoticeChannel,"*",whereMap,"id desc",0,0);
return getDataTable(list);
}
@ApiOperation("新增告警通知渠道")
@PreAuthorize("@ss.hasPermi('iot:alertNoticeChannel:add')")
@Log(title = "新增告警通知渠道", businessType = BusinessType.INSERT)
@PostMapping
public AjaxResult add(@RequestBody IotAlertNoticeChannel iotAlertNoticeChannel)
{
if (StringUtils.isEmpty(iotAlertNoticeChannel.getConfig()))
{
return AjaxResult.error("请输入渠道配置");
}
switch (iotAlertNoticeChannel.getType())
{
case 1:
WeiXinGongZhongHaoNoticeDto weiXinGongZhongHaoNoticeDto = JSONObject.parseObject(iotAlertNoticeChannel.getConfig(), WeiXinGongZhongHaoNoticeDto.class);
if(BeanUtil.isNotEmpty(weiXinGongZhongHaoNoticeDto))
{
return AjaxResult.error("渠道配置填写错误");
}
break;
}
return toAjax(publicService.insert(iotAlertNoticeChannel));
}
@ApiOperation("修改告警通知渠道")
@PreAuthorize("@ss.hasPermi('iot:alertNoticeChannel:edit')")
@Log(title = "topic权限控制", businessType = BusinessType.UPDATE)
@PutMapping
public AjaxResult edit(@RequestBody IotAlertNoticeChannel iotAlertNoticeChannel)
{
return toAjax(publicService.updateObject(iotAlertNoticeChannel,"id"));
}
@ApiOperation("删除告警通知渠道")
@PreAuthorize("@ss.hasPermi('iot:IotPermission:remove')")
@Log(title = "topic权限控制", businessType = BusinessType.DELETE)
@DeleteMapping("/{ids}")
public AjaxResult remove(@PathVariable Integer[] ids)
{
Long ct = publicService.selectCountBySql("SELECT COUNT(*) ct FROM iot_alert_user_notice WHERE CONCAT(',', channels, ',') LIKE '%,1,%'");
if(ct>0)
{
return AjaxResult.error("该告警通知渠道有用户告警通知配置使用，请先删除用户告警通知配置再删除此通道");
}
int i=0;
for (Integer id:ids)
{
i+=publicService.deleteObjectById(IotAlertNoticeChannel.class,id+"");
}
return toAjax(i);
}
}
... ...
... ... @@ -37,6 +37,10 @@
<groupId>org.quartz-scheduler</groupId>
<artifactId>quartz</artifactId>
</dependency>
<dependency>
<groupId>com.zhonglai</groupId>
<artifactId>weixin-api</artifactId>
</dependency>
</dependencies>
<build>
... ... @@ -62,7 +66,7 @@
生成的manifest中classpath的前缀，因为要把第三方jar放到lib目录下，所以classpath的前缀是lib/
-->
<classpathPrefix>lib/</classpathPrefix>
<mainClass>com.zhonglai.luhui.alarm.AlarmApplication</mainClass>
<mainClass>com.zhonglai.luhui.alarm.LhAlarmMain</mainClass>
</manifest>
</archive>
</configuration>
... ...
... ... @@ -3,6 +3,7 @@ package com.zhonglai.luhui.alarm;
import com.zhonglai.luhui.alarm.config.CachAlarmConfig;
import com.zhonglai.luhui.alarm.handle.CleanupTask;
import com.zhonglai.luhui.alarm.service.SendAlarmService;
import com.zhonglai.luhui.alarm.service.TimerAlarmService;
import com.zhonglai.luhui.alarm.service.TriggerAlarmService;
... ... @@ -18,6 +19,7 @@ public class LhAlarmMain {
//启动定时任务告警
TimerAlarmService.start();
//启动告警定时发送
SendAlarmService.start();
}
}
... ...
... ... @@ -34,7 +34,8 @@ public class IotAlertAlarm extends UpAlarmFactory<IotAlert> implements Serializa
@Override
void updateGenerateAlarm() {
switch (afterupAlarmDb.getAlertType())
int alertType = null==afterupAlarmDb.getAlertType()?beforeupAlarmDb.getAlertType():afterupAlarmDb.getAlertType();
switch (alertType)
{
case 1:
CachAlarmConfig.addAttributeAlarmRoute(afterupAlarmDb);
... ...
... ... @@ -41,7 +41,7 @@ public class IotDeviceAlarm extends UpAlarmFactory<IotDevice> implements Seriali
CachAlarmConfig.putDeviceProduct(afterupAlarmDb.getClient_id(), afterupAlarmDb.getProduct_id());
}
//通过变更的数据获得对应的告警规则
valueUp(beforeupAlarmDb.getClient_id(),CachAlarmConfig.getDeviceProduct(beforeupAlarmDb.getClient_id()),beforeupAlarmDb.getThings_model_value(),afterupAlarmDb.getThings_model_value());
valueUp(beforeupAlarmDb.getClient_id(),CachAlarmConfig.getDeviceProduct(beforeupAlarmDb.getClient_id()),beforeupAlarmDb.getThings_model_value(),afterupAlarmDb.getThings_model_value(),null,"主机"+afterupAlarmDb.getClient_id());
}
Object getNowValue(Object model_name,Integer type) {
... ...
... ... @@ -38,7 +38,7 @@ public class IotTerminalAlarm extends UpAlarmFactory<IotTerminal> implements Ser
@Override
public void updateGenerateAlarm() {
//通过变更的数据获得对应的告警规则
valueUp(beforeupAlarmDb.getId(), CachAlarmConfig.getDeviceProduct(beforeupAlarmDb.getId()),beforeupAlarmDb.getThings_model_value(),afterupAlarmDb.getThings_model_value());
valueUp(beforeupAlarmDb.getId(), CachAlarmConfig.getDeviceProduct(beforeupAlarmDb.getId()),beforeupAlarmDb.getThings_model_value(),afterupAlarmDb.getThings_model_value(),beforeupAlarmDb.getUser_info_id(),null==beforeupAlarmDb.getName()?beforeupAlarmDb.getId():beforeupAlarmDb.getName());
}
... ...
package com.zhonglai.luhui.alarm.clas;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONException;
import com.alibaba.fastjson.JSONObject;
import com.alibaba.otter.canal.protocol.CanalEntry;
import com.zhonglai.luhui.alarm.config.CachAlarmConfig;
import com.zhonglai.luhui.alarm.dto.*;
import com.zhonglai.luhui.alarm.service.TriggerAlarmService;
import com.zhonglai.luhui.alarm.util.GenericComparator;
import com.zhonglai.luhui.service.dao.util.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.ArrayList;
import java.util.List;
... ... @@ -15,7 +19,7 @@ import java.util.List;
* 告警工厂
*/
public abstract class UpAlarmFactory<T> {
private static final Logger logger = LoggerFactory.getLogger(UpAlarmFactory.class);
protected List<IotAlertLog> list = new ArrayList<>();
protected T beforeupAlarmDb;
... ... @@ -69,7 +73,7 @@ public abstract class UpAlarmFactory<T> {
}
protected void valueUp(String client_id,Integer product_id,String oldvalue,String newValue)
protected void valueUp(String client_id,Integer product_id,String oldvalue,String newValue,Integer user_id,String device_name)
{
if(StringUtils.isNotEmpty(newValue))
{
... ... @@ -81,50 +85,71 @@ public abstract class UpAlarmFactory<T> {
}
for (String key:newjson.keySet())
{
if(!newjson.get(key).equals(oldjson.get(key))) //新老不一致，触发告警
Object nv ;
Object ov ;
try {
JSONObject newObject = newjson.getJSONObject(key);
String type = newObject.getString("type");
nv = GenericComparator.convertToType(newObject.getString("saveView"),type);
ov = GenericComparator.convertToType(oldjson.getJSONObject(key).getString("saveView"),type);
}catch (JSONException e ) //如果不是json，就按照字符串来对比
{
nv = newjson.get(key);
ov = oldjson.get(key);
}
if(!nv.equals(ov)) //新老不一致，触发告警
{
//属性告警
List<IotAlert> listAttribute = CachAlarmConfig.getAttributeIotAlert(product_id,key);
if (null != listAttribute && listAttribute.size() != 0)
{
alarmAttribute(client_id,listAttribute,oldjson.get(key));
alarmAttribute(client_id,listAttribute,nv,user_id,device_name);
}
//触发告警
List<IotAlert> listTrigger = CachAlarmConfig.getTriggerIotAlert(product_id,key);
if (null != listTrigger && listTrigger.size() != 0)
{
alarmTrigger(client_id,listTrigger,oldjson.get(key));
alarmTrigger(client_id,listTrigger,nv,user_id,device_name);
}
}
}
}
}
protected void alarmAttribute(String client_id,List<IotAlert> listAttribute,Object newValue)
protected void alarmAttribute(String client_id,List<IotAlert> listAttribute,Object newValue,Integer user_id,String device_name)
{
for (IotAlert iotAlert:listAttribute)
{
if(null != user_id && null != iotAlert.getUser_id() && user_id - iotAlert.getUser_id()!=0) //不是指定用户的设备就不需要生产
{
continue;
}
AttributeTriggers attributeTriggers = JSONObject.parseObject(iotAlert.getTriggers(),AttributeTriggers.class);
String alrmname = attributeTriggers.getValueMapName().get(newValue);
if(StringUtils.isNotEmpty(alrmname))
{
IotAlertLog iotAlertLog = new IotAlertLog(iotAlert.getAlertId(),alrmname,iotAlert.getAlertLevel().intValue(),2,client_id,System.currentTimeMillis(),1,iotAlert.getCreateBy());
IotAlertLog iotAlertLog = new IotAlertLog(iotAlert.getAlertId(),alrmname,iotAlert.getAlertLevel().intValue(),2,client_id,System.currentTimeMillis(),1,iotAlert.getUser_id() ,device_name,null==iotAlert.getUser_id()?1:2);
list.add(iotAlertLog);
}
}
}
protected void alarmTrigger(String client_id,List<IotAlert> listAttribute,Object newValue)
protected void alarmTrigger(String client_id,List<IotAlert> listAttribute,Object newValue,Integer user_id,String device_name)
{
for (IotAlert iotAlert:listAttribute)
{
if(null != user_id && null != iotAlert.getUser_id() && user_id - iotAlert.getUser_id()!=0) //不是指定用户的设备就不需要生产
{
continue;
}
List<TriggerTriggers> attributeTriggersList = JSON.parseArray(iotAlert.getTriggers(),TriggerTriggers.class);
alarmTriggerTriggers(attributeTriggersList,newValue,iotAlert.getAlertId(),iotAlert.getAlertName(),iotAlert.getAlertLevel().intValue(),client_id,iotAlert.getCreateBy());
alarmTriggerTriggers(attributeTriggersList,newValue,iotAlert.getAlertId(),iotAlert.getAlertName(),iotAlert.getAlertLevel().intValue(),client_id,iotAlert.getUser_id(),device_name);
}
}
public void alarmTriggerTriggers( List<TriggerTriggers> attributeTriggersList,Object newValue,Long alert_id, String alert_name, Integer alert_level, String device_id,String userId)
public void alarmTriggerTriggers( List<TriggerTriggers> attributeTriggersList,Object newValue,Long alert_id, String alert_name, Integer alert_level, String device_id,Integer userId,String device_name )
{
if(null != attributeTriggersList && attributeTriggersList.size() != 0)
{
... ... @@ -163,7 +188,7 @@ public abstract class UpAlarmFactory<T> {
if(jieguo)
{
IotAlertLog iotAlertLog = new IotAlertLog(alert_id,alert_name,alert_level,2,device_id,System.currentTimeMillis(),1,userId);
IotAlertLog iotAlertLog = new IotAlertLog(alert_id,alert_name,alert_level,2,device_id,System.currentTimeMillis(),1,userId,device_name,null==userId?1:2);
list.add(iotAlertLog);
}
}
... ... @@ -171,7 +196,7 @@ public abstract class UpAlarmFactory<T> {
protected Object getValueFromJSON(String key,JSONObject jsonObject)
{
if(null != jsonObject && jsonObject.containsKey(jsonObject))
if(null != jsonObject && jsonObject.containsKey(key))
{
return jsonObject.getJSONObject(key).get("value");
}
... ...
... ... @@ -3,23 +3,14 @@ package com.zhonglai.luhui.alarm.config;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.zhonglai.luhui.alarm.dao.DbOperateUtil;
import com.zhonglai.luhui.alarm.dto.AttributeTriggers;
import com.zhonglai.luhui.alarm.dto.IotAlert;
import com.zhonglai.luhui.alarm.dto.TimerTriggers;
import com.zhonglai.luhui.alarm.dto.TriggerTriggers;
import com.zhonglai.luhui.alarm.handle.TimerAlarmJob;
import com.zhonglai.luhui.alarm.dto.*;
import com.zhonglai.luhui.alarm.service.TimerAlarmService;
import com.zhonglai.luhui.service.dao.util.StringUtils;
import org.quartz.*;
import org.quartz.impl.StdSchedulerFactory;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.sql.SQLException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.*;
/**
* 缓存的告警配置
... ... @@ -41,6 +32,15 @@ public class CachAlarmConfig {
private static Map<String, Integer> device_product = new HashMap<>();
/** 用户告警通知配置 */
private static Map<String, IotAlertUserNotice> user_alarm_config = new HashMap<>();
/** 用户告警通知渠道 */
private static Map<Integer, IotAlertNoticeChannel> user_alarm_notice_channel = new HashMap<>();
/** 用户微信小程序关系 */
private static Map<Integer, Map<Integer, Set<String>>> xcx_user_openid = new HashMap<>();
public static void loadConfig()
{
//加载属性告警配置
... ... @@ -54,6 +54,15 @@ public class CachAlarmConfig {
//加载用户告警配置
loadUserAlarmConfig();
//加载用户告警通知配置
loadUserAlarmNoticeConfig();
//加载用户告警通知渠道
loadUserAlarmNoticeChannel();
//加载用户微信小程序关系
loadXcxUserOpenid();
}
/**
... ... @@ -66,6 +75,7 @@ public class CachAlarmConfig {
{
for (IotAlert iotAlert:list)
{
iotAlert.setCreateBy(null);
addAttributeAlarmRoute(iotAlert);
}
}
... ... @@ -83,6 +93,7 @@ public class CachAlarmConfig {
{
for (IotAlert iotAlert:list)
{
iotAlert.setCreateBy(null);
addTriggerAlarmRoute(iotAlert);
}
}
... ... @@ -98,6 +109,7 @@ public class CachAlarmConfig {
{
for (IotAlert iotAlert:list)
{
iotAlert.setCreateBy(null);
TimerAlarmService.addTimerAlarm(iotAlert);
}
}
... ... @@ -134,6 +146,63 @@ public class CachAlarmConfig {
}
}
/**
* 加载用户告警通知配置
*/
private static void loadUserAlarmNoticeConfig()
{
List<IotAlertUserNotice> list = DbOperateUtil.getAlertUserNoticeList();
if(null != list && list.size() != 0)
{
for (IotAlertUserNotice iotAlertUserNotice:list)
{
user_alarm_config.put(iotAlertUserNotice.getUser_id()+"|"+iotAlertUserNotice.getType(),iotAlertUserNotice);
}
}
}
/**
* 加载用户告警通知渠道
*/
private static void loadUserAlarmNoticeChannel()
{
List<IotAlertNoticeChannel> list = DbOperateUtil.getAlertNoticeChannelList();
if(null != list && list.size() != 0)
{
for (IotAlertNoticeChannel iotAlertNoticeChannel:list)
{
user_alarm_notice_channel.put(iotAlertNoticeChannel.getId(),iotAlertNoticeChannel);
}
}
}
/**
* 加载用户微信小程序关系
*/
private static void loadXcxUserOpenid()
{
List<UserOpenid> list = DbOperateUtil.getUserOpenidList();
if(null != list && list.size() != 0)
{
for (UserOpenid userOpenid:list)
{
Map<Integer, Set<String>> mapset = xcx_user_openid.get(userOpenid.getXcx_config_id());
if(null == mapset)
{
mapset = new HashMap<>();
xcx_user_openid.put(userOpenid.getXcx_config_id(),mapset);
}
Set<String> set = mapset.get(userOpenid.getUser_id());
if(null == set)
{
set = new HashSet<>();
mapset.put(userOpenid.getUser_id(),set);
}
set.add(userOpenid.getOpenid());
}
}
}
/**
* 添加属性告警路由
... ... @@ -320,4 +389,47 @@ public class CachAlarmConfig {
{
alarmConfig.put(iotAlert.getAlertId(),iotAlert);
}
public static List<UserAlarmNoticeConfig> getUserAlarmNoticeConfig(Integer user_id,Integer getAlert_config_type,Long alert_id)
{
IotAlertUserNotice iotAlertUserNotice = user_alarm_config.get(user_id+"|"+getAlert_config_type);
if(null != iotAlertUserNotice && "*".equals(iotAlertUserNotice.getAlert_ids()) || (","+iotAlertUserNotice.getAlert_ids()+",").indexOf(","+alert_id+",")>=0) //*或者包含
{
return channelIdsToUserAlarmNoticeConfig(iotAlertUserNotice.getChannels());
}
return null;
}
private static List<UserAlarmNoticeConfig> channelIdsToUserAlarmNoticeConfig(String iot_alert_notice_channel_ids)
{
if(StringUtils.isNotEmpty(iot_alert_notice_channel_ids))
{
List<UserAlarmNoticeConfig> list = new ArrayList<>();
String[] ids = iot_alert_notice_channel_ids.split(",");
for (String id:ids)
{
Integer iid = Integer.parseInt(id);
IotAlertNoticeChannel iotAlertNoticeChannel = user_alarm_notice_channel.get(iid);
if(null != iotAlertNoticeChannel )
{
UserAlarmNoticeConfig userAlarmNoticeConfig = new UserAlarmNoticeConfig();
userAlarmNoticeConfig.setFromConfig(iotAlertNoticeChannel.getConfig());
userAlarmNoticeConfig.setChannelType(iotAlertNoticeChannel.getType());
list.add(userAlarmNoticeConfig);
}
}
return list;
}
return null;
}
public static Set<String> getUserXcxOpenid(Integer userId,Integer xcxConfigId)
{
if(xcx_user_openid.containsKey(xcxConfigId) && xcx_user_openid.get(xcxConfigId).containsKey(userId) )
{
return xcx_user_openid.get(xcxConfigId).get(userId);
}
return null;
}
}
... ...
package com.zhonglai.luhui.alarm.dao;
import com.zhonglai.luhui.alarm.dto.IotAlert;
import com.zhonglai.luhui.alarm.dto.IotAlertLog;
import com.zhonglai.luhui.alarm.dto.IotTerminal;
import com.zhonglai.luhui.alarm.dto.*;
import com.zhonglai.luhui.service.dao.BaseDao;
import com.zhonglai.luhui.service.dao.util.StringUtils;
import org.apache.commons.dbutils.handlers.ScalarHandler;
import java.sql.SQLException;
import java.util.List;
import java.util.stream.Collectors;
public class DbOperateUtil {
private static BaseDao baseDao = new BaseDao();
... ... @@ -36,13 +35,19 @@ public class DbOperateUtil {
return baseDao.insertList(list,null);
}
public static List<IotTerminal> getTimerAlarmIotTerminal(Long productId,String terminalids)
public static List<IotTerminal> getTimerAlarmIotTerminal(Long productId,String terminalids, Integer user_id)
{
String where = "";
if(null != user_id)
{
where = " and user_info_id = "+user_id;
}
if(StringUtils.isNotEmpty(terminalids))
{
return baseDao.findBysql("SELECT * FROM `iot_terminal` WHERE product_id=? AND id IN(?)",IotTerminal.class,productId,terminalids.split(","));
return baseDao.findBysql("SELECT * FROM `iot_terminal` WHERE product_id=? AND id IN(?)"+where,IotTerminal.class,productId,terminalids.split(","));
}
return baseDao.findBysql("SELECT * FROM `iot_terminal` WHERE product_id=?",IotTerminal.class,productId);
return baseDao.findBysql("SELECT * FROM `iot_terminal` WHERE product_id=?"+where,IotTerminal.class,productId);
}
public static List<IotAlertLog> getIotAlertLogList(Integer status,Integer limit)
... ... @@ -53,6 +58,21 @@ public class DbOperateUtil {
public static int updateIotAlertLogStatus(List<Long> ids,Integer status)
{
return baseDao.updateBySql("update iot_alert_log set `status`=? where id in(?)",status,ids);
return baseDao.updateBySql("update iot_alert_log set `status`="+status+" where alert_log_id in("+ids.stream().map(Object::toString).collect(Collectors.joining(", "))+")");
}
public static List<IotAlertUserNotice> getAlertUserNoticeList()
{
return baseDao.findBysql("SELECT * FROM `iot_alert_user_notice` where channels is not null and alert_ids is not null",IotAlertUserNotice.class);
}
public static List<IotAlertNoticeChannel> getAlertNoticeChannelList()
{
return baseDao.findBysql("SELECT * FROM `iot_alert_notice_channel`",IotAlertNoticeChannel.class);
}
public static List<UserOpenid> getUserOpenidList()
{
return baseDao.findBysql("SELECT * FROM `user_openid`",UserOpenid.class);
}
}
... ...
package com.zhonglai.luhui.alarm.dto;
import com.alibaba.otter.canal.protocol.CanalEntry;
import com.zhonglai.luhui.service.dao.util.StringUtils;
import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method;
... ... @@ -35,6 +36,15 @@ public class IotAlert {
private Integer status;
private String createBy;
private Integer user_id;
public Integer getUser_id() {
return user_id;
}
public void setUser_id(Integer user_id) {
this.user_id = user_id;
}
public String getCreateBy() {
return createBy;
... ... @@ -124,10 +134,41 @@ public class IotAlert {
IotAlert iotAlert = new IotAlert();
for (CanalEntry.Column column : columns)
{
try {
Method method = iotAlert.getClass().getMethod("set"+getName(column.getName()));
method.invoke(iotAlert, column.getValue());
} catch (NoSuchMethodException | IllegalAccessException | InvocationTargetException e) {
switch (column.getName())
{
case "alert_id":
iotAlert.setAlertId(Long.parseLong(column.getValue()));
break;
case "alert_name":
iotAlert.setAlertName(column.getValue());
break;
case "alert_level":
iotAlert.setAlertLevel(Long.parseLong(column.getValue()));
break;
case "alert_type":
iotAlert.setAlertType(Integer.parseInt(column.getValue()));
break;
case "product_id":
iotAlert.setProductId(Long.parseLong(column.getValue()));
break;
case "product_name":
iotAlert.setProductName(column.getValue());
break;
case "triggers":
iotAlert.setTriggers(column.getValue());
break;
case "actions":
iotAlert.setActions(column.getValue());
break;
case "status":
iotAlert.setStatus(Integer.parseInt(column.getValue()));
break;
case "create_by":
iotAlert.setCreateBy(column.getValue());
break;
case "user_id":
iotAlert.setUser_id(Integer.parseInt(column.getValue()));
break;
}
}
if(null == iotAlert.getAlertId())
... ...
... ... @@ -19,10 +19,12 @@ public class IotAlertLog
private Long create_time; // datetime DEFAULT NULL COMMENT '创建时间',
private Integer type; // tinyint DEFAULT NULL COMMENT '类型（1=告警，2=场景联动）',
private Integer user_id; //关联的用户id（0为系统生成）
private Integer alert_config_type; //告警配置类型(1系统配置，2用户配置)
private String device_name;
public IotAlertLog() {
}
public IotAlertLog(Long alert_id, String alert_name, Integer alert_level, Integer status, String device_id, Long create_time, Integer type,String user_id) {
public IotAlertLog(Long alert_id, String alert_name, Integer alert_level, Integer status, String device_id, Long create_time, Integer type,Integer user_id,String device_name,Integer alert_config_type) {
this.alert_id = alert_id;
this.alert_name = alert_name;
this.alert_level = alert_level;
... ... @@ -30,11 +32,25 @@ public class IotAlertLog
this.device_id = device_id;
this.create_time = create_time;
this.type = type;
if(StringUtils.isNotEmpty(user_id))
{
this.user_id = Integer.parseInt(user_id);
}
this.user_id = user_id;
this.device_name = device_name;
this.alert_config_type = alert_config_type;
}
public Integer getAlert_config_type() {
return alert_config_type;
}
public void setAlert_config_type(Integer alert_config_type) {
this.alert_config_type = alert_config_type;
}
public String getDevice_name() {
return device_name;
}
public void setDevice_name(String device_name) {
this.device_name = device_name;
}
public Integer getUser_id() {
... ...
package com.zhonglai.luhui.alarm.dto;
/**
* 告警通知渠道
*/
public class IotAlertNoticeChannel {
private Integer id; // int NOT NULL AUTO_INCREMENT COMMENT '主键',
private Integer type; // int NOT NULL COMMENT '渠道类型(1公众号，2小程序，3鱼儿乐app，4短信，5电话)',
private String config; // json NOT NULL COMMENT '渠道配置',
private String name; // varchar(50) NOT NULL COMMENT '名称',
public Integer getId() {
return id;
}
public void setId(Integer id) {
this.id = id;
}
public Integer getType() {
return type;
}
public void setType(Integer type) {
this.type = type;
}
public String getConfig() {
return config;
}
public void setConfig(String config) {
this.config = config;
}
public String getName() {
return name;
}
public void setName(String name) {
this.name = name;
}
}
... ...
package com.zhonglai.luhui.alarm.dto;
/**
* 用户告警通知配置
*/
public class IotAlertUserNotice {
private Integer id; // int NOT NULL AUTO_INCREMENT COMMENT '主键id',
private String channels; // varchar(100) DEFAULT NULL COMMENT '通知渠道(iot_alert_notice_channel表的id集合,多个英文逗号分割)',
private Integer type; // int DEFAULT NULL COMMENT '告警类型（1系统告警，2用户告警）',
private String alert_ids; // varchar(100) DEFAULT '*' COMMENT '告警id集合（关联iot_alert_log和iot_alert_user表的id,多个英文逗号分割，*表示所有）',
private Integer user_id; // int DEFAULT NULL COMMENT '用户id'
public Integer getId() {
return id;
}
public void setId(Integer id) {
this.id = id;
}
public Integer getType() {
return type;
}
public void setType(Integer type) {
this.type = type;
}
public Integer getUser_id() {
return user_id;
}
public void setUser_id(Integer user_id) {
this.user_id = user_id;
}
public String getChannels() {
return channels;
}
public void setChannels(String channels) {
this.channels = channels;
}
public String getAlert_ids() {
return alert_ids;
}
public void setAlert_ids(String alert_ids) {
this.alert_ids = alert_ids;
}
}
... ...
... ... @@ -72,7 +72,7 @@ public class IotDevice {
break;
}
}
if(null == iotDevice.getSummary() || null == iotDevice.getThings_model_value() || null == iotDevice.getClient_id())
if(null == iotDevice.getSummary() || null == iotDevice.getThings_model_value() || null == iotDevice.getClient_id() )
{
return null;
}
... ...
... ... @@ -20,6 +20,26 @@ public class IotTerminal {
private Integer product_id;
private Integer user_info_id;
private String name;
public String getName() {
return name;
}
public void setName(String name) {
this.name = name;
}
public Integer getUser_info_id() {
return user_info_id;
}
public void setUser_info_id(Integer user_info_id) {
this.user_info_id = user_info_id;
}
public String getDevice_id() {
return device_id;
}
... ... @@ -85,6 +105,12 @@ public class IotTerminal {
case "product_id":
iotTerminal.setProduct_id(null!=column.getValue()&&!"".equals(column.getValue())?Integer.parseInt(column.getValue()):null);
break;
case "user_info_id":
iotTerminal.setUser_info_id(null!=column.getValue()&&!"".equals(column.getValue())?Integer.parseInt(column.getValue()):null);
break;
case "name":
iotTerminal.setName(column.getValue());
break;
}
}
if(null == iotTerminal.getThings_model_value() || null == iotTerminal.getThings_model_config() || null == iotTerminal.getId())
... ...
package com.zhonglai.luhui.alarm.dto;
import java.util.List;
public class UserAlarmNoticeConfig {
private String fromConfig;
private Integer channelType;
public String getFromConfig() {
return fromConfig;
}
public void setFromConfig(String fromConfig) {
this.fromConfig = fromConfig;
}
public Integer getChannelType() {
return channelType;
}
public void setChannelType(Integer channelType) {
this.channelType = channelType;
}
}
... ...
package com.zhonglai.luhui.alarm.dto;
public class UserOpenid {
private Integer user_id; // int DEFAULT NULL COMMENT '用户id',
private String openid; // varchar(50) DEFAULT NULL COMMENT '微信用户的唯一标识',
private Integer xcx_config_id; // int DEFAULT NULL COMMENT '微信平台的配置参数表id',
public Integer getUser_id() {
return user_id;
}
public void setUser_id(Integer user_id) {
this.user_id = user_id;
}
public String getOpenid() {
return openid;
}
public void setOpenid(String openid) {
this.openid = openid;
}
public Integer getXcx_config_id() {
return xcx_config_id;
}
public void setXcx_config_id(Integer xcx_config_id) {
this.xcx_config_id = xcx_config_id;
}
}
... ...
... ... @@ -27,17 +27,18 @@ public class TimerAlarmJob implements Job {
List<TriggerTriggers> triggersList = timerTriggers.getTriggersList();
String alertName = iotAlert.getAlertName();
Long productId = iotAlert.getProductId();
Integer user_id = iotAlert.getUser_id();
String terminalids = timerTriggers.getTerminalids();
List<IotAlertLog> list = new ArrayList<>();
List<IotTerminal> iotTerminallist = DbOperateUtil.getTimerAlarmIotTerminal(productId,terminalids);
List<IotTerminal> iotTerminallist = DbOperateUtil.getTimerAlarmIotTerminal(productId,terminalids,user_id);
if (null != iotTerminallist && iotTerminallist.size() != 0 )
{
for (IotTerminal iotTerminal:iotTerminallist)
{
IotTerminalAlarm iotTerminalAlarm = new IotTerminalAlarm(iotTerminal,iotTerminal);
iotTerminalAlarm.alarmTriggerTriggers(triggersList,iotTerminal.getThings_model_value(),iotAlert.getAlertId(),alertName,iotAlert.getAlertLevel().intValue(),iotTerminal.getId(),iotAlert.getCreateBy());
iotTerminalAlarm.alarmTriggerTriggers(triggersList,iotTerminal.getThings_model_value(),iotAlert.getAlertId(),alertName,iotAlert.getAlertLevel().intValue(),iotTerminal.getId(),iotAlert.getUser_id(), iotTerminal.getName());
List<IotAlertLog> serviceDtolist = iotTerminalAlarm.getList();
if(null != serviceDtolist && serviceDtolist.size()!=0)
{
... ...
package com.zhonglai.luhui.alarm.notice;
import com.alibaba.fastjson.JSONObject;
import com.zhonglai.luhui.alarm.config.CachAlarmConfig;
import com.zhonglai.luhui.alarm.dto.IotAlertLog;
import com.zhonglai.luhui.alarm.dto.UserAlarmNoticeConfig;
import com.zhonglai.luhui.alarm.notice.dto.WeiXinGongZhongHaoNoticeDto;
import com.zhonglai.luhui.alarm.notice.impl.WeiXinGongZhongHaoNoticeImpl;
import java.util.List;
public interface NoticeFactory {
boolean send(IotAlertLog iotAlertLog);
static boolean notice(IotAlertLog iotAlertLog)
{
//获取通知配置
List<UserAlarmNoticeConfig> userAlarmNoticeConfigList = CachAlarmConfig.getUserAlarmNoticeConfig(iotAlertLog.getUser_id(),iotAlertLog.getAlert_config_type(),iotAlertLog.getAlert_id());
if(null != userAlarmNoticeConfigList && userAlarmNoticeConfigList.size()!=0) //有配置告警通知的才发送
{
boolean rb = true;
for (UserAlarmNoticeConfig userAlarmNoticeConfig:userAlarmNoticeConfigList)
{
NoticeFactory noticeFactory = null;
//通过配置的通知渠道，分发消息
switch (userAlarmNoticeConfig.getChannelType()) //渠道类型(1公众号，2小程序，3鱼儿乐app，4短信，5电话)
{
case 1:
WeiXinGongZhongHaoNoticeDto weiXinGongZhongHaoNoticeDto = JSONObject.parseObject(userAlarmNoticeConfig.getFromConfig(),WeiXinGongZhongHaoNoticeDto.class);
noticeFactory = new WeiXinGongZhongHaoNoticeImpl(iotAlertLog.getUser_id(),weiXinGongZhongHaoNoticeDto);
break;
case 2:
break;
case 3:
break;
case 4:
break;
case 5:
break;
}
if(null != noticeFactory)
{
rb=rb&&noticeFactory.send(iotAlertLog);
}
}
return rb;
}else {
return false;
}
}
static boolean request(IotAlertLog iotAlertLog)
{
//先通知
boolean rb = notice(iotAlertLog);
//确认到达
return rb;
}
}
... ...
package com.zhonglai.luhui.alarm.notice.dto;
/**
* 公众号通知的配置
*/
public class WeiXinGongZhongHaoNoticeDto {
private Integer xcxConfigId;
private String xcxappid;
private String xcxsecret;
private String gzhappid;
private String gzhsecret;
private String templateId;
private String redirectUrl;
public String getGzhsecret() {
return gzhsecret;
}
public void setGzhsecret(String gzhsecret) {
this.gzhsecret = gzhsecret;
}
public Integer getXcxConfigId() {
return xcxConfigId;
}
public void setXcxConfigId(Integer xcxConfigId) {
this.xcxConfigId = xcxConfigId;
}
public String getXcxappid() {
return xcxappid;
}
public void setXcxappid(String xcxappid) {
this.xcxappid = xcxappid;
}
public String getXcxsecret() {
return xcxsecret;
}
public void setXcxsecret(String xcxsecret) {
this.xcxsecret = xcxsecret;
}
public String getGzhappid() {
return gzhappid;
}
public void setGzhappid(String gzhappid) {
this.gzhappid = gzhappid;
}
public String getTemplateId() {
return templateId;
}
public void setTemplateId(String templateId) {
this.templateId = templateId;
}
public String getRedirectUrl() {
return redirectUrl;
}
public void setRedirectUrl(String redirectUrl) {
this.redirectUrl = redirectUrl;
}
}
... ...
package com.zhonglai.luhui.alarm.notice.dto.wxUniformSend;
public class Miniprogram {
private String appid; //小程序appid
private String pagepath; //
public String getAppid() {
return appid;
}
public void setAppid(String appid) {
this.appid = appid;
}
public String getPagepath() {
return pagepath;
}
public void setPagepath(String pagepath) {
this.pagepath = pagepath;
}
}
... ...
package com.zhonglai.luhui.alarm.notice.dto.wxUniformSend;
import weixin.popular.bean.message.templatemessage.TemplateMessageItem;
import java.util.LinkedHashMap;
public class MpTemplateMsg {
private String appid; //公众号appid，要求与小程序有绑定且同主体
private String template_id; //公众号模板id
private String url; //公众号模板消息所要跳转的url
private Miniprogram miniprogram; //公众号模板消息所要跳转的小程序，小程序的必须与公众号具有绑定关系
private LinkedHashMap<String, TemplateMessageItem> data; //公众号模板消息的数据
public String getAppid() {
return appid;
}
public void setAppid(String appid) {
this.appid = appid;
}
public String getTemplate_id() {
return template_id;
}
public void setTemplate_id(String template_id) {
this.template_id = template_id;
}
public String getUrl() {
return url;
}
public void setUrl(String url) {
this.url = url;
}
public Miniprogram getMiniprogram() {
return miniprogram;
}
public void setMiniprogram(Miniprogram miniprogram) {
this.miniprogram = miniprogram;
}
public LinkedHashMap<String, TemplateMessageItem> getData() {
return data;
}
public void setData(LinkedHashMap<String, TemplateMessageItem> data) {
this.data = data;
}
}
... ...
package com.zhonglai.luhui.alarm.notice.dto.wxUniformSend;
import weixin.popular.bean.message.templatemessage.TemplateMessageItem;
import java.util.LinkedHashMap;
public class WeappTemplateMsg {
private String touser;
private String template_id;
private String page;
private String form_id;
private String emphasis_keyword;
private LinkedHashMap<String, TemplateMessageItem> data;
public String getTouser() {
return touser;
}
public void setTouser(String touser) {
this.touser = touser;
}
public String getTemplate_id() {
return template_id;
}
public void setTemplate_id(String template_id) {
this.template_id = template_id;
}
public String getPage() {
return page;
}
public void setPage(String page) {
this.page = page;
}
public String getForm_id() {
return form_id;
}
public void setForm_id(String form_id) {
this.form_id = form_id;
}
public String getEmphasis_keyword() {
return emphasis_keyword;
}
public void setEmphasis_keyword(String emphasis_keyword) {
this.emphasis_keyword = emphasis_keyword;
}
public LinkedHashMap<String, TemplateMessageItem> getData() {
return data;
}
public void setData(LinkedHashMap<String, TemplateMessageItem> data) {
this.data = data;
}
}
... ...
package com.zhonglai.luhui.alarm.notice.dto.wxUniformSend;
import com.alibaba.fastjson.JSONObject;
import com.zhonglai.luhui.alarm.notice.dto.wxUniformSend.Miniprogram;
import com.zhonglai.luhui.alarm.notice.dto.wxUniformSend.MpTemplateMsg;
import com.zhonglai.luhui.alarm.notice.dto.wxUniformSend.WxUniformSendMessage;
import org.apache.http.Header;
import org.apache.http.client.methods.HttpUriRequest;
import org.apache.http.client.methods.RequestBuilder;
import org.apache.http.entity.ContentType;
import org.apache.http.entity.StringEntity;
import org.apache.http.message.BasicHeader;
import weixin.popular.api.MessageAPI;
import weixin.popular.api.TokenAPI;
import weixin.popular.bean.message.templatemessage.TemplateMessage;
import weixin.popular.bean.message.templatemessage.TemplateMessageItem;
import weixin.popular.bean.message.templatemessage.TemplateMessageResult;
import weixin.popular.bean.token.Token;
import weixin.popular.client.LocalHttpClient;
import weixin.popular.util.JsonUtil;
import java.nio.charset.Charset;
import java.util.HashMap;
import java.util.LinkedHashMap;
import java.util.Map;
public class WeiXinMessage {
private static Map<String,Token> tokenMap = new HashMap<>();
/**
* 发送公众号模版消息
*
* @param openid 用户标识
* @param templateId 模版id
* @param redirectUrl 模版消息重定向地址
* @param parameters 模版消息参数
*/
public static void sendTemplateMsg(String appid, String secret,String openid, String templateId, String redirectUrl, Map<String, String> parameters) {
TemplateMessage templateMessage = new TemplateMessage();
templateMessage.setTemplate_id(templateId);
templateMessage.setTouser(openid);
templateMessage.setData(parametersToData(parameters));
templateMessage.setUrl(redirectUrl);
String defaultToken = getWechatServerToken(appid,secret);
TemplateMessageResult templateMessageResult = MessageAPI.messageTemplateSend(defaultToken, templateMessage);
}
/**
* 下发小程序和公众号统一的服务消息
*
* @param openid 用户标识
* @param templateId 模版id
* @param redirectUrl 模版消息重定向地址
* @param parameters 模版消息参数
*/
public static void sendUniformMessage(String gzhappid,String xcxappid, String secret,String openid, String templateId, String redirectUrl, Map<String, String> parameters) {
WxUniformSendMessage wxUniformSendMessage = new WxUniformSendMessage();
wxUniformSendMessage.setTouser(openid);
// wxUniformSendMessage.setWeapp_template_msg(parametersToData(parameters));
MpTemplateMsg mpTemplateMsg = new MpTemplateMsg();
mpTemplateMsg.setAppid(gzhappid);
mpTemplateMsg.setData(parametersToData(parameters));
Miniprogram miniprogram = new Miniprogram();
miniprogram.setAppid(xcxappid);
miniprogram.setPagepath(redirectUrl);
mpTemplateMsg.setMiniprogram(miniprogram);
mpTemplateMsg.setTemplate_id(templateId);
// mpTemplateMsg.setUrl(redirectUrl);
wxUniformSendMessage.setMp_template_msg(mpTemplateMsg);
String defaultToken = getWechatServerToken(xcxappid,secret);
TemplateMessageResult templateMessageResult = uniform_send(defaultToken, wxUniformSendMessage);
}
private static LinkedHashMap<String, TemplateMessageItem> parametersToData(Map<String, String> parameters)
{
LinkedHashMap<String, TemplateMessageItem> msgMap = new LinkedHashMap<>();
if (parameters != null && parameters.size() > 0) {
for (String key : parameters.keySet()) {
msgMap.put(key, new TemplateMessageItem(parameters.get(key), ""));
}
}
return msgMap;
}
private static Header jsonHeader = new BasicHeader("Content-Type", ContentType.APPLICATION_JSON.toString());
/**
* 下发小程序和公众号统一的服务消息
* @param access_token
* @return
*/
public static TemplateMessageResult uniform_send(String access_token, WxUniformSendMessage wxUniformSendMessage) {
String messageJson = JsonUtil.toJSONString(wxUniformSendMessage);
HttpUriRequest httpUriRequest = RequestBuilder.post().setHeader(jsonHeader).setUri("https://api.weixin.qq.com/cgi-bin/message/wxopen/template/uniform_send").addParameter("access_token", access_token).setEntity(new StringEntity(messageJson, Charset.forName("utf-8"))).build();
return LocalHttpClient.executeJsonResult(httpUriRequest, TemplateMessageResult.class);
}
/**
* 发送模板消息
* @return
*/
public static TemplateMessageResult messageTemplateSend(String appid, String secret, JSONObject jsonObject) {
String messageJson = jsonObject.toJSONString();
HttpUriRequest httpUriRequest = RequestBuilder.post().setHeader(jsonHeader).setUri("https://api.weixin.qq.com/cgi-bin/message/template/send").addParameter("access_token", getWechatServerToken(appid,secret)).setEntity(new StringEntity(messageJson, Charset.forName("utf-8"))).build();
return LocalHttpClient.executeJsonResult(httpUriRequest, TemplateMessageResult.class);
}
/**
* 获取服务号的token
*
* @return
*/
public static String getWechatServerToken(String appid, String secret) {
Token token = tokenMap.get(appid);
int time =new Long(System.currentTimeMillis()).intValue();
if(null != token && token.getExpires_in()-time>0)
{
return token.getAccess_token();
}
token = TokenAPI.token(appid,secret);
token.setExpires_in(time+token.getExpires_in()-600); //提前10分钟
return token.getAccess_token();
}
}
... ...
package com.zhonglai.luhui.alarm.notice.dto.wxUniformSend;
public class WxUniformSendMessage {
private String touser;
private WeappTemplateMsg weapp_template_msg;
private MpTemplateMsg mp_template_msg;
public String getTouser() {
return touser;
}
public void setTouser(String touser) {
this.touser = touser;
}
public WeappTemplateMsg getWeapp_template_msg() {
return weapp_template_msg;
}
public void setWeapp_template_msg(WeappTemplateMsg weapp_template_msg) {
this.weapp_template_msg = weapp_template_msg;
}
public MpTemplateMsg getMp_template_msg() {
return mp_template_msg;
}
public void setMp_template_msg(MpTemplateMsg mp_template_msg) {
this.mp_template_msg = mp_template_msg;
}
}
... ...
package com.zhonglai.luhui.alarm.notice.impl;
import com.alibaba.fastjson.JSONObject;
import com.zhonglai.luhui.alarm.config.CachAlarmConfig;
import com.zhonglai.luhui.alarm.dto.IotAlertLog;
import com.zhonglai.luhui.alarm.notice.NoticeFactory;
import com.zhonglai.luhui.alarm.notice.dto.WeiXinGongZhongHaoNoticeDto;
import com.zhonglai.luhui.alarm.notice.dto.wxUniformSend.WeiXinMessage;
import org.springframework.beans.factory.annotation.Value;
import weixin.popular.api.MessageAPI;
import weixin.popular.bean.message.templatemessage.Miniprogram;
import weixin.popular.bean.message.templatemessage.TemplateMessage;
import weixin.popular.bean.message.templatemessage.TemplateMessageItem;
import java.text.SimpleDateFormat;
import java.util.*;
/**
* 微信公众号
*/
public class WeiXinGongZhongHaoNoticeImpl implements NoticeFactory {
private Set<String> toOpenIds;
private WeiXinGongZhongHaoNoticeDto weiXinGongZhongHaoNoticeDto;
public WeiXinGongZhongHaoNoticeImpl(Integer userId,WeiXinGongZhongHaoNoticeDto weiXinGongZhongHaoNoticeDto)
{
this.weiXinGongZhongHaoNoticeDto = weiXinGongZhongHaoNoticeDto;
toOpenIds = CachAlarmConfig.getUserXcxOpenid(userId,weiXinGongZhongHaoNoticeDto.getXcxConfigId());
}
@Override
public boolean send(IotAlertLog iotAlertLog) {
if(null != toOpenIds && toOpenIds.size() != 0)
{
return sendMessage(iotAlertLog,toOpenIds.toArray(new String[toOpenIds.size()]));
}
return false;
}
/**
* 设备告警消息
* @param os openid
*/
public boolean sendMessage(IotAlertLog iotAlertLog, String ... os)
{
JSONObject data = new JSONObject();
data.put("thing23",new org.json.JSONObject().put("value",iotAlertLog.getDevice_name()));
data.put("character_string11",new org.json.JSONObject().put("value",iotAlertLog.getDevice_id()));
data.put("time3",new org.json.JSONObject().put("value",new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(new Date(iotAlertLog.getCreate_time()))));
data.put("thing33",new org.json.JSONObject().put("value",iotAlertLog.getAlert_name()));
Miniprogram miniprogram = new Miniprogram();
miniprogram.setAppid(weiXinGongZhongHaoNoticeDto.getXcxappid());
miniprogram.setPagepath(weiXinGongZhongHaoNoticeDto.getRedirectUrl());
for(String o:os)
{
JSONObject templateMessage = new JSONObject();
templateMessage.put("touser",o);
templateMessage.put("template_id",weiXinGongZhongHaoNoticeDto.getTemplateId());
templateMessage.put("data",data);
templateMessage.put("miniprogram",miniprogram);
WeiXinMessage.messageTemplateSend(weiXinGongZhongHaoNoticeDto.getGzhappid(),weiXinGongZhongHaoNoticeDto.getGzhsecret(),templateMessage);
}
return true;
}
}
... ...
package com.zhonglai.luhui.alarm.service;
import com.zhonglai.luhui.alarm.config.CachAlarmConfig;
import com.zhonglai.luhui.alarm.dao.DbOperateUtil;
import com.zhonglai.luhui.alarm.dto.IotAlertLog;
import com.zhonglai.luhui.alarm.notice.NoticeFactory;
import com.zhonglai.luhui.alarm.util.ThreadPoolUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
/**
* 发送告警业务
*/
public class SendAlarmService {
private static final Logger logger = LoggerFactory.getLogger(SendAlarmService.class);
public static void start()
{
/**该方法相对起始时间点以固定频率调用指定的任务（fixed-rate任务)。当把任务提交到线程池并延迟initialDelay时间后开始执行任务command。
* 然后从initialDelay+period时间点再次执行，而后在initialDelay+2*period时间点再次执行，循环往复，直到抛出异常或者调用了任务的cancel方法取消了任务，
* 或者关闭了线程池。
*/
ThreadPoolUtil.executor.scheduleWithFixedDelay(() -> runSend(),0,15, TimeUnit.SECONDS);
}
private static void runSend()
{
List<IotAlertLog> list = DbOperateUtil.getIotAlertLogList(2,100); //未处理的告警
if(null != list && list.size()!=0)
{
... ... @@ -20,31 +39,48 @@ public class SendAlarmService {
for (IotAlertLog iotAlertLog:list)
{
if(null == iotAlertLog.getUser_id())
{
notids.add(iotAlertLog.getAlert_log_id());
continue;
}
switch (iotAlertLog.getAlert_level())
{
case 1: //轻微问题
notids.add(iotAlertLog.getAlert_log_id());
break;
case 2: //提醒通知
ids.add(iotAlertLog.getAlert_log_id());
notice(iotAlertLog);
if(NoticeFactory.notice(iotAlertLog))
{
ids.add(iotAlertLog.getAlert_log_id());
}else {
notids.add(iotAlertLog.getAlert_log_id());
}
break;
case 3: //严重警告
ids.add(iotAlertLog.getAlert_log_id());
if(NoticeFactory.request(iotAlertLog))
{
ids.add(iotAlertLog.getAlert_log_id());
}else{
notids.add(iotAlertLog.getAlert_log_id());
}
break;
default:
notids.add(iotAlertLog.getAlert_log_id());
break;
}
}
DbOperateUtil.updateIotAlertLogStatus(ids,3);
DbOperateUtil.updateIotAlertLogStatus(notids,1);
if(ids.size()!=0)
{
DbOperateUtil.updateIotAlertLogStatus(ids,3);
}
if(notids.size()!=0)
{
DbOperateUtil.updateIotAlertLogStatus(notids,1);
}
}
}
public static void notice(IotAlertLog iotAlertLog)
{
logger.info("发送任务执行完成");
}
}
... ...
... ... @@ -7,10 +7,12 @@ import com.alibaba.otter.canal.protocol.CanalEntry.Entry;
import com.alibaba.otter.canal.protocol.CanalEntry.EntryType;
import com.alibaba.otter.canal.protocol.CanalEntry.RowChange;
import com.alibaba.otter.canal.protocol.CanalEntry.RowData;
import com.google.protobuf.InvalidProtocolBufferException;
import com.zhonglai.luhui.alarm.clas.UpAlarmFactory;
import com.zhonglai.luhui.alarm.config.CachAlarmConfig;
import com.zhonglai.luhui.alarm.dao.DbOperateUtil;
import com.zhonglai.luhui.alarm.dto.IotAlertLog;
import com.zhonglai.luhui.alarm.util.ThreadPoolUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
... ... @@ -24,13 +26,10 @@ import java.util.concurrent.*;
*/
public class TriggerAlarmService {
private static final Logger logger = LoggerFactory.getLogger(TriggerAlarmService.class);
private static int poolSize = 100; // 线程池大小
private static ScheduledExecutorService executor = Executors.newScheduledThreadPool(poolSize);
private static Semaphore semaphore = new Semaphore(poolSize); // 信号量，许可数等于线程池大小
// 创建Canal连接器
private static CanalConnector connector = CanalConnectors.newSingleConnector(
new InetSocketAddress("8.129.82.37", 11111),
new InetSocketAddress("127.0.0.1", 11111),
"example",
"",
""
... ... @@ -38,18 +37,21 @@ public class TriggerAlarmService {
public static void start() {
try {
connect();
} catch (InterruptedException e) {
throw new RuntimeException(e);
}
// 异步获取数据
executor.scheduleWithFixedDelay(new Runnable() {
ThreadPoolUtil.executor.scheduleWithFixedDelay(new Runnable() {
@Override
public void run() {
try {
semaphore.acquire(); // 获取信号量许可，如果没有可用许可则阻塞等待
connect();
handle();
}catch (Exception e)
{
logger.info("触发告警业务异常",e);
}finally {
semaphore.release(); // 释放信号量许可
}
}
... ... @@ -62,37 +64,40 @@ public class TriggerAlarmService {
connector.connect();
// 订阅指定的表（这里以database.table为例）
connector.subscribe("mqtt_broker\\.(iot_device|iot_terminal|iot_alert)");
}
while (true) {
semaphore.acquire(); // 获取信号量许可，如果没有可用许可则阻塞等待
executor.schedule(() -> {
// 获取指定数量的数据
Message message = connector.getWithoutAck(100);
long batchId = message.getId();
int size = message.getEntries().size();
if (batchId != -1 && size > 0) {
processEntries(message.getEntries());
}
// 确认消息被消费
connector.ack(batchId);
},0, TimeUnit.SECONDS);
private static void handle() throws InterruptedException {
while (ThreadPoolUtil.executor.getPoolSize()-ThreadPoolUtil.executor.getActiveCount()>0)
{
logger.info("总线程数{}，已使用线程数{}，剩下线程数{}",ThreadPoolUtil.executor.getPoolSize(),ThreadPoolUtil.executor.getActiveCount(),ThreadPoolUtil.executor.getPoolSize()-ThreadPoolUtil.executor.getActiveCount());
// 获取指定数量的数据
Message message = connector.getWithoutAck(100);
long batchId = message.getId();
int size = message.getEntries().size();
logger.info("拉取的数据库同步数据量{}",size);
if (batchId != -1 && size > 0) {
ThreadPoolUtil.executor.schedule(() -> {
try {
processEntries(message.getEntries());
}catch (Exception e)
{
logger.info("数据处理异常"+message.toString(),e);
}
},0, TimeUnit.SECONDS);
}
// 确认消息被消费
connector.ack(batchId);
}
}
private static void processEntries(List<Entry> entries) {
private static void processEntries(List<Entry> entries) throws InvalidProtocolBufferException {
List<IotAlertLog> list = new ArrayList<>();
for (Entry entry : entries) {
if (entry.getEntryType() == EntryType.ROWDATA) {
RowChange rowChange = null;
try {
rowChange = RowChange.parseFrom(entry.getStoreValue());
} catch (Exception e) {
throw new RuntimeException("ERROR ## parser of eromanga-event has an error, data:" + entry.toString(), e);
}
RowChange rowChange = RowChange.parseFrom(entry.getStoreValue());;
// 获取数据库名和表名
String dbName = entry.getHeader().getSchemaName();
String tableName = entry.getHeader().getTableName();
... ... @@ -124,16 +129,6 @@ public class TriggerAlarmService {
{
// 断开连接
connector.disconnect();
// 关闭调度器
executor.shutdown();
try {
if (!executor.awaitTermination(5, TimeUnit.SECONDS)) {
executor.shutdownNow();
}
} catch (InterruptedException e) {
executor.shutdownNow();
Thread.currentThread().interrupt();
}
ThreadPoolUtil.close();
}
}
... ...
package com.zhonglai.luhui.alarm.util;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import java.math.BigDecimal;
import static com.alibaba.fastjson.JSON.parseArray;
public class GenericComparator {
public static <T> boolean compare(T a, T b, String equation) {
if (a == null || b == null || equation == null) {
throw new IllegalArgumentException("Arguments cannot be null");
return false;
}
if (!a.getClass().equals(b.getClass())) {
throw new IllegalArgumentException("Arguments must be of the same type");
b = (T) convertToType(b, a.getClass());
}
Comparable<T> compA = (Comparable<T>) a;
... ... @@ -32,13 +40,63 @@ public class GenericComparator {
}
}
public static Object convertToType(Object obj, String targetType) {
switch (targetType.toUpperCase())
{
case "INTEGER":
return new BigDecimal(obj.toString());
case "DECIMAL":
return new BigDecimal(obj.toString());
case "BOOL":
return Boolean.valueOf(obj.toString());
case "ARRAY":
return obj.toString();
default:
return obj;
}
}
private static Object convertToType(Object obj, Class<?> targetType) {
if (targetType.isInstance(obj)) {
return obj;
}
if (targetType == Integer.class) {
return Integer.valueOf(obj.toString());
} else if (targetType == Double.class) {
return Double.valueOf(obj.toString());
} else if (targetType == Float.class) {
return Float.valueOf(obj.toString());
} else if (targetType == Long.class) {
return Long.valueOf(obj.toString());
} else if (targetType == Short.class) {
return Short.valueOf(obj.toString());
} else if (targetType == Byte.class) {
return Byte.valueOf(obj.toString());
} else if (targetType == Boolean.class) {
return Boolean.valueOf(obj.toString());
} else if (targetType == Character.class) {
return obj.toString().charAt(0);
}else if(targetType == BigDecimal.class)
{
return new BigDecimal(obj.toString());
}
else {
return obj.toString();
}
}
public static void main(String[] args) {
// 示例测试
System.out.println(compare(5, 3, ">")); // true
System.out.println(compare(5.5, 3.3, "==")); // false
System.out.println(compare("apple", "banana", "<")); // true
System.out.println(compare(5, 5, "==")); // true
System.out.println(compare("a", "a", "==")); // true
System.out.println(compare(5.5, 5.3, "<=")); // false
// System.out.println(compare(5, "3", ">")); // true
// System.out.println(compare(5.5, 3.3, "==")); // false
// System.out.println(compare("apple", "banana", "<")); // true
// System.out.println(compare(5, 5, "==")); // true
// System.out.println(compare("a", "a", "==")); // true
// System.out.println(compare(5.5, 5.3, "<=")); // false
System.out.println(compare(new BigDecimal(5.3),"5.3", "<")); // false
}
}
... ...
package com.zhonglai.luhui.alarm.util;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
public class ThreadPoolUtil {
private static int poolSize = 100; // 线程池大小
public static ScheduledThreadPoolExecutor executor = new ScheduledThreadPoolExecutor (poolSize);
public static void close()
{
// 关闭调度器
executor.shutdown();
try {
if (!executor.awaitTermination(5, TimeUnit.SECONDS)) {
executor.shutdownNow();
}
} catch (InterruptedException e) {
executor.shutdownNow();
Thread.currentThread().interrupt();
}
}
}
... ...
... ... @@ -69,6 +69,10 @@
<groupId>org.aspectj</groupId>
<artifactId>aspectjrt</artifactId>
</dependency>
<dependency>
<groupId>com.zhonglai</groupId>
<artifactId>weixin-api</artifactId>
</dependency>
</dependencies>
<build>
... ...
... ... @@ -8,14 +8,17 @@ import com.ruoyi.common.core.domain.Message;
import com.ruoyi.common.enums.BusinessType;
import com.ruoyi.common.utils.GsonConstructor;
import com.zhonglai.luhui.action.BaseController;
import com.zhonglai.luhui.device.domain.IotDevice;
import com.zhonglai.luhui.device.dto.CommandType;
import com.zhonglai.luhui.device.dto.DeviceCommand;
import com.zhonglai.luhui.device.service.IIotDeviceService;
import com.zhonglai.luhui.rocketmq.service.RocketMqService;
import io.swagger.annotations.Api;
import io.swagger.annotations.ApiImplicitParam;
import io.swagger.annotations.ApiImplicitParams;
import io.swagger.annotations.ApiOperation;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.security.access.prepost.PreAuthorize;
import org.springframework.util.StreamUtils;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.PostMapping;
... ... @@ -25,12 +28,15 @@ import org.springframework.web.bind.annotation.RestController;
import javax.servlet.http.HttpServletRequest;
import java.io.IOException;
@Api(tags = "设备控制")
@Api(tags = "控制器")
@RestController
@RequestMapping("/iot/controlDevice")
public class ControlDeviceConreoller extends BaseController {
@Autowired
private RocketMqService rocketMqService;
@Autowired
private IIotDeviceService iotDeviceService;
@ApiOperation(value = "写指令",notes = "body参数描述：\r\n" +
"{\n" +
" \"0\":{\n" +
... ... @@ -64,7 +70,45 @@ public class ControlDeviceConreoller extends BaseController {
deviceCommand.setDeviceId(deviceId);
deviceCommand.setCommandType(CommandType.write);
deviceCommand.setData(GsonConstructor.get().fromJson(body,JsonObject.class));
return control(deviceCommand);
return deviceControl(deviceCommand);
}
@ApiOperation(value = "读指令",notes = "body参数描述：\r\n" +
"{\n" +
" \"0\":{\n" +
" \"rssi\": -43,\n" +
" \"firmware_version\": 1.2,\n" +
" \"longitude\": 0,\n" +
" \"latitude\": 0,\n" +
" \"summary\": {\n" +
" \t \"name\": \"wumei-smart\",\n" +
" \t \"chip\": \"esp8266\",\n" +
" \t \"author\": \"kerwincui\",\n" +
" \t \"version\": 1.2,\n" +
" \t \"createTime\": \"2022-06-06\"\n" +
" }\n" +
" },\n" +
" \"1\":{\n" +
" \"id1\":\"value1\",\n" +
" \"id2\":\"value2\",\n" +
" \"id3\":\"value3\"\n" +
" }\n" +
"}")
@ApiImplicitParams({
@ApiImplicitParam(value = "网关id",name = "deviceId"),
})
@Log(title = "读指令", businessType = BusinessType.WRITE)
@PostMapping("/read/{deviceId}")
public AjaxResult read(@PathVariable String deviceId, HttpServletRequest request) throws IOException {
byte[] bodyBytes = StreamUtils.copyToByteArray(request.getInputStream());
String body = new String(bodyBytes, request.getCharacterEncoding());
DeviceCommand deviceCommand = new DeviceCommand();
deviceCommand.setDeviceId(deviceId);
deviceCommand.setCommandType(CommandType.write);
deviceCommand.setData(GsonConstructor.get().fromJson(body,JsonObject.class));
return deviceControl(deviceCommand);
}
@ApiOperation("更新缓存模型")
... ... @@ -77,7 +121,7 @@ public class ControlDeviceConreoller extends BaseController {
JsonObject jsonObject = new JsonObject();
jsonObject.addProperty("product_id",product_id);
deviceCommand.setData(jsonObject);
return control(deviceCommand);
return sysControl(deviceCommand);
}
@ApiOperation("更新缓存翻译模型")
... ... @@ -90,7 +134,7 @@ public class ControlDeviceConreoller extends BaseController {
JsonObject jsonObject = new JsonObject();
jsonObject.addProperty("product_id",product_id);
deviceCommand.setData(jsonObject);
return control(deviceCommand);
return sysControl(deviceCommand);
}
@ApiOperation("清除网关缓存")
... ... @@ -101,7 +145,7 @@ public class ControlDeviceConreoller extends BaseController {
DeviceCommand deviceCommand = new DeviceCommand();
deviceCommand.setDeviceId(deviceId);
deviceCommand.setCommandType(CommandType.cleanDeviceHost);
return control(deviceCommand);
return sysControl(deviceCommand);
}
@ApiOperation("更新终端缓存")
... ... @@ -118,13 +162,68 @@ public class ControlDeviceConreoller extends BaseController {
JsonObject jsonObject = new JsonObject();
jsonObject.addProperty("sensor_number",sensor_number);
deviceCommand.setData(jsonObject);
return control(deviceCommand);
return sysControl(deviceCommand);
}
@ApiOperation("添加订阅")
@ApiImplicitParams({
@ApiImplicitParam(value = "监听服务器的ip",name = "ip"),
@ApiImplicitParam(value = "产品集合",name = "product_ids"),
})
@Log(title = "添加订阅", businessType = BusinessType.CAHE)
@PostMapping("/addSubscribe")
public AjaxResult addSubscribe(String product_ids,String ip) {
DeviceCommand deviceCommand = new DeviceCommand();
deviceCommand.setCommandType(CommandType.addSubscribe);
JsonObject jsonObject = new JsonObject();
jsonObject.addProperty("product_ids",product_ids);
jsonObject.addProperty("ip",ip);
deviceCommand.setData(jsonObject);
return sysControl(deviceCommand);
}
@ApiOperation(value = "通知",notes = "body参数描述：\r\n" +
"{\n" +
" \"1\":{\n" +
" \"id1\":\"value1\",\n" +
" \"id2\":\"value2\",\n" +
" \"id3\":\"value3\"\n" +
" }\n" +
"}")
@ApiImplicitParams({
@ApiImplicitParam(value = "网关id",name = "deviceId"),
})
@Log(title = "通知", businessType = BusinessType.CAHE)
@PostMapping("/notice/{deviceId}")
public AjaxResult notice(@PathVariable String deviceId, HttpServletRequest request) throws IOException {
byte[] bodyBytes = StreamUtils.copyToByteArray(request.getInputStream());
String body = new String(bodyBytes, request.getCharacterEncoding());
DeviceCommand deviceCommand = new DeviceCommand();
deviceCommand.setDeviceId(deviceId);
deviceCommand.setCommandType(CommandType.notice);
deviceCommand.setData(GsonConstructor.get().fromJson(body,JsonObject.class));
return deviceControl(deviceCommand);
}
private AjaxResult deviceControl(DeviceCommand deviceCommand)
{
IotDevice iotDevice = iotDeviceService.selectIotDeviceByClient_id(deviceCommand.getDeviceId());
if(null == iotDevice || iotDevice.getStatus()!=3)
{
return AjaxResult.error("设备不存在或者不在线");
}
Message message = rocketMqService.send("deviceCommandListen",GsonConstructor.get().toJson(deviceCommand).getBytes(),iotDevice.getOperation_token());
if(message.getCode()==1)
{
return new AjaxResult(HttpStatus.SUCCESS, message.getMessage(), message.getData());
}
return new AjaxResult(HttpStatus.ERROR, message.getMessage(), message.getData());
}
private AjaxResult control( DeviceCommand deviceCommand)
private AjaxResult sysControl(DeviceCommand deviceCommand)
{
Message message = rocketMqService.send("deviceCommandListen",GsonConstructor.get().toJson(deviceCommand).getBytes());
Message message = rocketMqService.send("deviceCommandListen",GsonConstructor.get().toJson(deviceCommand).getBytes(),"SysCommand");
if(message.getCode()==1)
{
return new AjaxResult(HttpStatus.SUCCESS, message.getMessage(), message.getData());
... ...
... ... @@ -39,6 +39,8 @@ public class IotTerminalController extends BaseController
private IIotTerminalService iotTerminalService;
@Autowired
private IUserTerminalGroupRelationService userTerminalGroupRelationService;
@Autowired
private PublicService publicService;
@ApiOperation("查询终端列表")
@GetMapping("/list")
public TableDataInfo list(IotTerminal iotTerminal)
... ... @@ -91,6 +93,33 @@ public class IotTerminalController extends BaseController
return toAjax(iotTerminalService.updateIotTerminal(iotTerminal));
}
@ApiOperation("删除")
@ApiImplicitParams({
@ApiImplicitParam(value = "终端id",name = "id"),
})
@Log(title = "删除", businessType = BusinessType.DELETE)
@Transactional
@PostMapping("/del")
public AjaxResult del(String id)
{
IotTerminal iotTerminal = iotTerminalService.selectIotTerminalById(id);
if(null == iotTerminal)
{
return AjaxResult.error("终端不存在");
}
Integer userid= SecurityUtils.getUserId().intValue();
if(null!=iotTerminal.getUser_info_id() && iotTerminal.getUser_info_id()-userid!=0)
{
return AjaxResult.error("该设备不是你的无法删除");
}
publicService.updateBySql("update iot_terminal set user_info_id=null where id='"+id+"'");
publicService.updateBySql("DELETE FROM `user_terminal_group_relation` WHERE user_info_id="+userid+" AND iot_terminal_id='"+id+"'");
return toAjax(iotTerminalService.updateIotTerminal(iotTerminal));
}
@ApiOperation("分组终端")
@ApiImplicitParams({
@ApiImplicitParam(value = "分组id",name = "iot_terminal_group_id"),
... ... @@ -124,7 +153,7 @@ public class IotTerminalController extends BaseController
{
return AjaxResult.error("请给所有设备都设置名称");
}
iotTerminalService.batchUpName(ids,names);
iotTerminalService.batchUpName(ids,names,userid);
return toAjax( userTerminalGroupRelationService.groupTerminal(iot_terminal_group_id,ids,userid));
}
}
... ...
package com.zhonglai.luhui.api.controller.user;
import com.alibaba.fastjson.JSONObject;
import com.ruoyi.common.annotation.Log;
import com.ruoyi.common.core.domain.AjaxResult;
import com.ruoyi.common.enums.BusinessType;
import com.ruoyi.common.utils.StringUtils;
import com.ruoyi.system.domain.user.UserOpenid;
import com.zhonglai.luhui.action.BaseController;
import com.zhonglai.luhui.dao.service.PublicService;
import com.zhonglai.luhui.device.domain.IotAlertNoticeChannel;
import com.zhonglai.luhui.device.domain.IotAlertUserNotice;
import com.zhonglai.luhui.device.dto.WeiXinGongZhongHaoNoticeDto;
import com.zhonglai.luhui.security.utils.SecurityUtils;
import io.swagger.annotations.Api;
import io.swagger.annotations.ApiImplicitParam;
import io.swagger.annotations.ApiImplicitParams;
import io.swagger.annotations.ApiOperation;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.web.bind.annotation.*;
import weixin.popular.api.SnsAPI;
import weixin.popular.bean.sns.SnsToken;
import javax.servlet.http.HttpServletRequest;
import java.io.UnsupportedEncodingException;
import java.net.URLEncoder;
@Api(tags = "用户管理")
@RestController
@RequestMapping("/user/userInfo")
public class UserInfoController extends BaseController {
@Autowired
private PublicService publicService;
@ApiOperation("提交小程序或者微信公众号通知服务")
@ApiImplicitParams({
@ApiImplicitParam(value = "渠道id（管理员提供）",name = "channel_id"),
@ApiImplicitParam(value = "授权获取到的code",name = "code"),
})
@Log(title = "提交小程序或者微信公众号通知服务", businessType = BusinessType.INSERT)
@Transactional
@PostMapping(value = "subXcxNoticeService/{channel_id}")
public AjaxResult subXcxNoticeService(@PathVariable Integer channel_id, String code)
{
//检查渠道是否存在
IotAlertNoticeChannel iotAlertNoticeChannel = publicService.getObject(IotAlertNoticeChannel.class,"id",channel_id+"");
if(null == iotAlertNoticeChannel)
{
return AjaxResult.error("通知渠道不存在");
}
if(iotAlertNoticeChannel.getType()!=1)
{
return AjaxResult.error("该渠道不是公众号通知渠道");
}
WeiXinGongZhongHaoNoticeDto weiXinGongZhongHaoNoticeDto = JSONObject.parseObject(iotAlertNoticeChannel.getConfig(),WeiXinGongZhongHaoNoticeDto.class);
Integer user_id = SecurityUtils.getUserId().intValue();
SnsToken snsToken = SnsAPI.oauth2AccessToken(weiXinGongZhongHaoNoticeDto.getGzhappid(), weiXinGongZhongHaoNoticeDto.getGzhsecret(),code);
if(StringUtils.isEmpty(snsToken.getOpenid()))
{
return AjaxResult.error("公众号授权失败");
}
//更新用户openid表
UserOpenid userOpenid = new UserOpenid();
userOpenid.setUser_id(user_id);
userOpenid.setOpenid(snsToken.getOpenid());
userOpenid.setXcx_config_id(weiXinGongZhongHaoNoticeDto.getXcxConfigId());
Long ct = publicService.getObjectListTotle(userOpenid,null);
if(0 == ct)
{
publicService.insert(userOpenid);
}
//更新用户通知
IotAlertUserNotice iotAlertUserNotice = new IotAlertUserNotice();
iotAlertUserNotice.setUser_id(user_id);
ct = publicService.getObjectListTotle(userOpenid,null);
if(0 < ct)
{
return AjaxResult.success("您已经开通过通知服务了,请到通知服务编辑页面进行编辑");
}
IotAlertUserNotice sysIotAlertUserNotice = new IotAlertUserNotice();
sysIotAlertUserNotice.setChannels(channel_id+"");
sysIotAlertUserNotice.setType(1);
sysIotAlertUserNotice.setUser_id(user_id);
publicService.insert(sysIotAlertUserNotice);
IotAlertUserNotice userIotAlertUserNotice = new IotAlertUserNotice();
userIotAlertUserNotice.setChannels(channel_id+"");
userIotAlertUserNotice.setType(2);
userIotAlertUserNotice.setUser_id(user_id);
publicService.insert(userIotAlertUserNotice);
return AjaxResult.success();
}
@ApiOperation("生成告警授权连接")
@RequestMapping(value = "getAlarmOauth2URL/{channel_id}", method = RequestMethod.GET)
public AjaxResult getAlarmOauth2URL(@PathVariable Integer channel_id, @RequestParam(value="redirect_uri") String redirect_uri,String state)
{
//检查渠道是否存在
IotAlertNoticeChannel iotAlertNoticeChannel = publicService.getObject(IotAlertNoticeChannel.class,"id",channel_id+"");
if(null == iotAlertNoticeChannel)
{
return AjaxResult.error("通知渠道不存在");
}
if(iotAlertNoticeChannel.getType()!=1)
{
return AjaxResult.error("该渠道不是公众号通知渠道");
}
WeiXinGongZhongHaoNoticeDto weiXinGongZhongHaoNoticeDto = JSONObject.parseObject(iotAlertNoticeChannel.getConfig(),WeiXinGongZhongHaoNoticeDto.class);
String authorizeUrl = SnsAPI.connectOauth2Authorize(weiXinGongZhongHaoNoticeDto.getGzhappid(), redirect_uri, true, state);
return AjaxResult.success("",authorizeUrl);
}
}
... ...
... ... @@ -27,6 +27,7 @@ public class ProtocolParserServiceImpl implements ProtocolParserFactory<JsonObje
case "ONLINE":
return new Online().analysisPayload(payload);
case "ALL_POST":
topic.setAllup(true);
return new AllPost().analysisPayload(payload);
case "ADD_POST":
return new AddPost().analysisPayload(payload);
... ...
... ... @@ -62,6 +62,7 @@ public class ProtocolParserAndPurificationFactory<T> {
ProtocolPurificationModel protocolPurificationModel = protocolPurificationFactory.purification(iotProduct.getId(),topic,analysisResult);
protocolPurificationModel.setNoticeClien(analysisResult.isNoticeClien());
protocolPurificationModel.setNoticeMessage(analysisResult.getNoticeMessage());
protocolPurificationModel.setAllUp(topic.isAllup());
return protocolPurificationModel;
}
... ...
... ... @@ -26,6 +26,7 @@ public class DefaultProtocolParserFactoryImpl implements ProtocolParserFactory<b
case "ONLINE":
return new Online().analysisPayload(payload);
case "ALL_POST":
topic.setAllup(true);
return new AllPost().analysisPayload(payload);
case "ADD_POST":
return new AddPost().analysisPayload(payload);
... ...
package com.zhonglai.luhui.device.protocol.factory.config;
import com.google.gson.JsonObject;
import com.ruoyi.common.utils.GsonConstructor;
import com.zhonglai.luhui.device.protocol.factory.dto.ParserDeviceHostDto;
import net.jodah.expiringmap.ExpirationListener;
import net.jodah.expiringmap.ExpirationPolicy;
... ... @@ -8,6 +9,8 @@ import net.jodah.expiringmap.ExpiringMap;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.TimeUnit;
/**
... ... @@ -15,11 +18,11 @@ import java.util.concurrent.TimeUnit;
*/
public class HttpNoticeCach {
private static Logger log = LoggerFactory.getLogger(HttpNoticeCach.class);
private static ExpiringMap<String, JsonObject> noticeUpdataData = ExpiringMap.builder().maxSize(20000).expiration(60, TimeUnit.SECONDS)
private static ExpiringMap<String, Map<String,Object>> noticeUpdataData = ExpiringMap.builder().maxSize(20000).expiration(60, TimeUnit.SECONDS)
.variableExpiration()
.asyncExpirationListener(new ExpirationListener<String, JsonObject>() {
.asyncExpirationListener(new ExpirationListener<String, Map<String,Object>>() {
@Override
public void expired(String s, JsonObject parserDeviceHostDto) {
public void expired(String s, Map<String,Object> parserDeviceHostDto) {
log.info("超时清除>>>>>>>：{} ",s);
}
})
... ... @@ -29,17 +32,17 @@ public class HttpNoticeCach {
{
if(noticeUpdataData.containsKey(parserDeviceHostDto.getId()))
{
noticeUpdataData.put(parserDeviceHostDto.getId(),jsonObject);
noticeUpdataData.put(parserDeviceHostDto.getId(), GsonConstructor.get().fromJson(jsonObject.toString(), HashMap.class));
}else {
noticeUpdataData.put(parserDeviceHostDto.getId(),jsonObject,parserDeviceHostDto.getDevice_life()*3,TimeUnit.SECONDS);
noticeUpdataData.put(parserDeviceHostDto.getId(),GsonConstructor.get().fromJson(jsonObject.toString(), HashMap.class),parserDeviceHostDto.getDevice_life()*3,TimeUnit.SECONDS);
}
}
public static JsonObject consumer(String imei)
public static Map<String,Object> consumer(String imei)
{
if(noticeUpdataData.containsKey(imei))
{
JsonObject jsonObject = noticeUpdataData.get(imei);
Map<String,Object> jsonObject = noticeUpdataData.get(imei);
noticeUpdataData.remove(imei);
return jsonObject;
}
... ...
... ... @@ -171,8 +171,6 @@ public class PluginsClassLoader extends URLClassLoader {
Class clazz = classMap.get(classname);
if (null != clazz)
{
System.out.println("接口的类加载器："+tClass.getClassLoader());
System.out.println("实现类的类加载器："+clazz.getClassLoader());
try {
if (tClass.isAssignableFrom(clazz) && !clazz.equals(tClass)) {
return tClass.cast(clazz.getDeclaredConstructor().newInstance());
... ...
package com.zhonglai.luhui.device.protocol.factory.control;
import cn.hutool.extra.spring.SpringUtil;
import com.google.gson.JsonObject;
import com.ruoyi.common.utils.GsonConstructor;
import com.zhonglai.luhui.device.analysis.comm.clien.ClienConnection;
import com.zhonglai.luhui.device.analysis.comm.clien.impl.ClienConnectionImpl;
import com.zhonglai.luhui.device.analysis.comm.config.SysParameter;
import com.zhonglai.luhui.device.analysis.comm.dto.ApiClientRePlyDto;
import com.zhonglai.luhui.device.analysis.comm.dto.TerminalClientRePlyDto;
import com.zhonglai.luhui.device.analysis.comm.factory.Topic;
... ... @@ -23,17 +25,27 @@ import com.zhonglai.luhui.device.protocol.factory.service.PersistenceDBService;
import net.jodah.expiringmap.ExpirationListener;
import net.jodah.expiringmap.ExpirationPolicy;
import net.jodah.expiringmap.ExpiringMap;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.spring.annotation.MessageModel;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.annotation.SelectorType;
import org.apache.rocketmq.spring.core.RocketMQPushConsumerLifecycleListener;
import org.apache.rocketmq.spring.core.RocketMQReplyListener;
import org.apache.rocketmq.spring.support.DefaultRocketMQListenerContainer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;
import javax.annotation.PostConstruct;
import java.util.Arrays;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.TimeUnit;
... ... @@ -42,8 +54,8 @@ import java.util.concurrent.TimeUnit;
* 设备指令监听服务
*/
@Service
@RocketMQMessageListener(consumerGroup = "deviceCommand", topic = "deviceCommandListen",messageModel = MessageModel.BROADCASTING)
public class DeviceCommandListenService implements RocketMQReplyListener<MessageExt, Message> {
@RocketMQMessageListener(consumerGroup = "deviceCommand", topic = "deviceCommandListen",selectorType = SelectorType.TAG,selectorExpression = "${rocketmq.operationToken}",messageModel = MessageModel.BROADCASTING)
public class DeviceCommandListenService implements RocketMQReplyListener<MessageExt, Message>, RocketMQPushConsumerLifecycleListener {
private static final Logger log = LoggerFactory.getLogger(DeviceCommandListenService.class);
private static ExpiringMap<String, ClienConnection> clienConnectionMap = ExpiringMap.builder().maxSize(20000).expiration(15, TimeUnit.SECONDS)
... ... @@ -63,15 +75,11 @@ public class DeviceCommandListenService implements RocketMQReplyListener<Message
@Autowired
private ClienNoticeServiceFactory clienNoticeServiceFactory;
@Autowired
private IotThingsModelService iotThingsModelService;
@Autowired
private MqttSubscribeService mqttSubscribeService;
@Value("${mqtt.client.operationTime}")
private long operationTime; //客户端操作时间
private String selectorExpression;
private DefaultMQPushConsumer defaultMQPushConsumer;
@Override
public Message onMessage(MessageExt messageExt) {
log.info("监听到消息{}",messageExt);
... ... @@ -81,88 +89,60 @@ public class DeviceCommandListenService implements RocketMQReplyListener<Message
DeviceCommand deviceCommand = GsonConstructor.get().fromJson(str, DeviceCommand.class);
if (deviceCommand.getCommandType().equals(CommandType.read) || deviceCommand.getCommandType().equals(CommandType.write) || deviceCommand.getCommandType().equals(CommandType.notice))
ParserDeviceHostDto parserDeviceHostDto = DeviceCach.getDeviceHost(deviceCommand.getDeviceId());
if(null == parserDeviceHostDto)
{
ParserDeviceHostDto parserDeviceHostDto = DeviceCach.getDeviceHost(deviceCommand.getDeviceId());
if(null == parserDeviceHostDto)
{
return new Message(MessageCode.DEFAULT_FAIL_CODE,"设备连接丢失，请稍后尝试");
}
return new Message(MessageCode.DEFAULT_FAIL_CODE,"设备连接丢失，请稍后尝试");
}
if (clienConnectionMap.containsKey(deviceCommand.getDeviceId()))
{
return new Message(MessageCode.DEFAULT_FAIL_CODE,"有其他人在控制设备，请稍后重试！");
}
if (clienConnectionMap.containsKey(deviceCommand.getDeviceId()))
{
return new Message(MessageCode.DEFAULT_FAIL_CODE,"有其他人在控制设备，请稍后重试！");
}
String classname = persistenceDBService.getClassnameFromIotProtocolClassId(parserDeviceHostDto.getIotProduct().getAnalysis_clas());
try {
ProtocolParserFactory protocolParserFactory = PluginsClassLoader.getJarClass(ProtocolParserFactory.class,classname);
String controlCls = classname.replace("analysis."+protocolParserFactory.getClass().getSimpleName(),"control.DeviceCommandListenServiceImpl");
DeviceCommandServiceFactory deviceCommandServiceFactory = PluginsClassLoader.getJarClass(DeviceCommandServiceFactory.class,controlCls);
if(null != deviceCommandServiceFactory)
String classname = persistenceDBService.getClassnameFromIotProtocolClassId(parserDeviceHostDto.getIotProduct().getAnalysis_clas());
try {
ProtocolParserFactory protocolParserFactory = PluginsClassLoader.getJarClass(ProtocolParserFactory.class,classname);
String controlCls = classname.replace("analysis."+protocolParserFactory.getClass().getSimpleName(),"control.DeviceCommandListenServiceImpl");
DeviceCommandServiceFactory deviceCommandServiceFactory = PluginsClassLoader.getJarClass(DeviceCommandServiceFactory.class,controlCls);
if(null != deviceCommandServiceFactory)
{
switch (deviceCommand.getCommandType())
{
switch (deviceCommand.getCommandType())
{
case read:
NoticeMessageDto noticeMessageDomain = deviceCommandServiceFactory.read(deviceCommand.getDeviceId(), deviceCommand.getData());
if(null == noticeMessageDomain)
{
return new Message(MessageCode.DEFAULT_FAIL_CODE,"该设备不支持读取功能");
}
return sendMessage(noticeMessageDomain);
case write:
noticeMessageDomain = deviceCommandServiceFactory.write(deviceCommand.getDeviceId(), deviceCommand.getData());
if(null == noticeMessageDomain)
{
return new Message(MessageCode.DEFAULT_FAIL_CODE,"该设备不支持写入功能");
}
return sendMessage(noticeMessageDomain);
case notice:
if(deviceCommandServiceFactory.notice(deviceCommand.getDeviceId(), deviceCommand.getData()))
{
return new Message(MessageCode.DEFAULT_SUCCESS_CODE,"指令发送成功");
}else {
return new Message(MessageCode.DEFAULT_FAIL_CODE,"指令发送失败");
}
default:
return new Message(MessageCode.DEFAULT_FAIL_CODE,"指令类型不存在，请联系管理员");
}
}else {
return new Message(MessageCode.DEFAULT_FAIL_CODE,"该设备的控制服务异常，请联系管理员");
case read:
NoticeMessageDto noticeMessageDomain = deviceCommandServiceFactory.read(deviceCommand.getDeviceId(), deviceCommand.getData());
if(null == noticeMessageDomain)
{
return new Message(MessageCode.DEFAULT_FAIL_CODE,"该设备不支持读取功能");
}
return sendMessage(noticeMessageDomain);
case write:
noticeMessageDomain = deviceCommandServiceFactory.write(deviceCommand.getDeviceId(), deviceCommand.getData());
if(null == noticeMessageDomain)
{
return new Message(MessageCode.DEFAULT_FAIL_CODE,"该设备不支持写入功能");
}
return sendMessage(noticeMessageDomain);
case notice:
if(deviceCommandServiceFactory.notice(deviceCommand.getDeviceId(), deviceCommand.getData()))
{
return new Message(MessageCode.DEFAULT_SUCCESS_CODE,"指令发送成功");
}else {
return new Message(MessageCode.DEFAULT_FAIL_CODE,"指令发送失败");
}
default:
return new Message(MessageCode.DEFAULT_FAIL_CODE,"指令类型不存在，请联系管理员");
}
} catch (InterruptedException e) {
log.error("消息发送失败",e);
return new Message(MessageCode.DEFAULT_FAIL_CODE,"等待通知错误");
}
}else{
switch (deviceCommand.getCommandType())
{
case cleanDeviceHost:
DeviceCach.cleanDeviceHost(deviceCommand.getDeviceId());
return new Message(MessageCode.DEFAULT_SUCCESS_CODE,"指令发送成功");
case cleanDeviceInfo:
DeviceCach.cleanDeviceInfo(deviceCommand.getDeviceId()+"_"+deviceCommand.getData().get("sensor_number").getAsString());
return new Message(MessageCode.DEFAULT_SUCCESS_CODE,"指令发送成功");
case upIotThingsModel:
iotThingsModelService.upIotThingsModel(deviceCommand.getData().get("product_id").getAsInt());
return new Message(MessageCode.DEFAULT_SUCCESS_CODE,"指令发送成功");
case upIotThingsModelTranslate:
iotThingsModelService.upIotThingsModelTranslate(deviceCommand.getData().get("product_id").getAsInt());
return new Message(MessageCode.DEFAULT_SUCCESS_CODE,"指令发送成功");
case addSubscribe:
Set<Integer> okProduct = mqttSubscribeService.assignIpAddSubscribe(deviceCommand.getData().get("ip").getAsString(),deviceCommand.getData().get("product_ids").getAsString());
if (null == okProduct || okProduct.size() ==0)
{
return new Message(MessageCode.DEFAULT_FAIL_CODE,"订阅未成功请检查原因");
}
return new Message(MessageCode.DEFAULT_SUCCESS_CODE,"订阅成功", Arrays.toString(okProduct.toArray()));
default:
return new Message(MessageCode.DEFAULT_FAIL_CODE,"指令类型不存在，请联系管理员");
}else {
return new Message(MessageCode.DEFAULT_FAIL_CODE,"该设备的控制服务异常，请联系管理员");
}
} catch (InterruptedException e) {
log.error("消息发送失败",e);
return new Message(MessageCode.DEFAULT_FAIL_CODE,"等待通知错误");
}
}
public Message sendMessage( NoticeMessageDto noticeMessageDomain) throws InterruptedException {
Topic topic = noticeMessageDomain.getTopic();
... ... @@ -196,19 +176,6 @@ public class DeviceCommandListenService implements RocketMQReplyListener<Message
return clienConnectionMap.containsKey(clientid);
}
private void log(Topic topic,JsonObject jsonObject)
{
// AddPostDto addPostDto = new AddPostDto();
// addPostDto.setData(JSON.parseObject(jsonObject.toString()));
// addPostDto.setIotTerminalList(new ArrayList<>());
// addPostDto.setDeviceSensorDataList(new ArrayList<>());
// addPostDto.setLogDeviceOperationList(new ArrayList<>());
// businessDataUpdateService.updataDta(BusinessDataUpdateService.Type.ADD,topic,addPostDto,"远程控制",true);
// cacheServiceImpl.updateCache(topic,addPostDto);
// addPostDto.setLogDeviceOperationList(logDeviceOperationList);
// dataPersistenceService.persistence(topic,addPostDto);
}
/**
* 通知给api操作端
* @param clientid
... ... @@ -230,4 +197,29 @@ public class DeviceCommandListenService implements RocketMQReplyListener<Message
log.info("结束通知{}",clientid);
}
@Override
public void prepareStart(DefaultMQPushConsumer consumer) {
//设置当前实例的名称
consumer.setInstanceName(this.getClass().getSimpleName());
this.defaultMQPushConsumer = consumer;
}
public String getSelectorExpression()
{
if(null == selectorExpression)
{
String[] beannames = SpringUtil.getBeanNamesForType(DefaultRocketMQListenerContainer.class);
if(null != beannames)
{
for (String beanname:beannames)
{
DefaultRocketMQListenerContainer container = SpringUtil.getBean(beanname);
if (container.getConsumer().equals(this.defaultMQPushConsumer)) {
selectorExpression = container.getSelectorExpression();
}
}
}
}
return selectorExpression;
}
}
... ...
package com.zhonglai.luhui.device.protocol.factory.control;
import com.ruoyi.common.utils.GsonConstructor;
import com.zhonglai.luhui.device.analysis.dto.Message;
import com.zhonglai.luhui.device.analysis.dto.MessageCode;
import com.zhonglai.luhui.device.protocol.factory.config.DeviceCach;
import com.zhonglai.luhui.device.protocol.factory.dto.DeviceCommand;
import com.zhonglai.luhui.device.protocol.factory.service.IotThingsModelService;
import com.zhonglai.luhui.device.protocol.factory.service.MqttSubscribeService;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.spring.annotation.MessageModel;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.annotation.SelectorType;
import org.apache.rocketmq.spring.core.RocketMQPushConsumerLifecycleListener;
import org.apache.rocketmq.spring.core.RocketMQReplyListener;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import java.util.Arrays;
import java.util.Set;
@Service
@RocketMQMessageListener(consumerGroup = "deviceCommand", topic = "deviceCommandListen",selectorType = SelectorType.TAG,selectorExpression = "SysCommand",messageModel = MessageModel.BROADCASTING)
public class SysCommandListenService implements RocketMQReplyListener<MessageExt, Message> , RocketMQPushConsumerLifecycleListener {
private static final Logger log = LoggerFactory.getLogger(DeviceCommandListenService.class);
@Autowired
private IotThingsModelService iotThingsModelService;
@Autowired
private MqttSubscribeService mqttSubscribeService;
@Override
public Message onMessage(MessageExt messageExt) {
log.info("监听到消息{}",messageExt);
String str = new String(messageExt.getBody());
log.info("消息body{}",str);
DeviceCommand deviceCommand = GsonConstructor.get().fromJson(str, DeviceCommand.class);
switch (deviceCommand.getCommandType())
{
case cleanDeviceHost:
DeviceCach.cleanDeviceHost(deviceCommand.getDeviceId());
return new Message(MessageCode.DEFAULT_SUCCESS_CODE,"指令发送成功");
case cleanDeviceInfo:
DeviceCach.cleanDeviceInfo(deviceCommand.getDeviceId()+"_"+deviceCommand.getData().get("sensor_number").getAsString());
return new Message(MessageCode.DEFAULT_SUCCESS_CODE,"指令发送成功");
case upIotThingsModel:
iotThingsModelService.upIotThingsModel(deviceCommand.getData().get("product_id").getAsInt());
return new Message(MessageCode.DEFAULT_SUCCESS_CODE,"指令发送成功");
case upIotThingsModelTranslate:
iotThingsModelService.upIotThingsModelTranslate(deviceCommand.getData().get("product_id").getAsInt());
return new Message(MessageCode.DEFAULT_SUCCESS_CODE,"指令发送成功");
case addSubscribe:
Set<Integer> okProduct = mqttSubscribeService.assignIpAddSubscribe(deviceCommand.getData().get("ip").getAsString(),deviceCommand.getData().get("product_ids").getAsString());
if (null == okProduct || okProduct.size() ==0)
{
return new Message(MessageCode.DEFAULT_FAIL_CODE,"订阅未成功请检查原因");
}
return new Message(MessageCode.DEFAULT_SUCCESS_CODE,"订阅成功", Arrays.toString(okProduct.toArray()));
default:
return new Message(MessageCode.DEFAULT_FAIL_CODE,"指令类型不存在，请联系管理员");
}
}
@Override
public void prepareStart(DefaultMQPushConsumer consumer) {
//设置当前实例的名称
consumer.setInstanceName(this.getClass().getSimpleName());
}
}
... ...
... ... @@ -26,6 +26,16 @@ public class ProtocolPurificationModel {
private List<DeviceSensorData> deviceSensorDataList = new ArrayList<>();
private List<LogDeviceOperation> logDeviceOperationList = new ArrayList<>();
private boolean isAllUp = false;
public boolean isAllUp() {
return isAllUp;
}
public void setAllUp(boolean allUp) {
isAllUp = allUp;
}
public ParserDeviceHostDto getParserDeviceHostDto() {
return parserDeviceHostDto;
}
... ...
... ... @@ -48,13 +48,13 @@ public abstract class BaseCallback<T> {
/**
* 数据处理的工作流
*/
protected void messageArrived(String imei,String s,T payload) throws ClassNotFoundException, InstantiationException, IllegalAccessException {
protected ProtocolPurificationModel messageArrived(String imei,String s,T payload) throws ClassNotFoundException, InstantiationException, IllegalAccessException {
//判断网关是否存在
ParserDeviceHostDto oldparserDeviceHostDto = persistenceDBService.getOldParserDeviceHostDto(imei);
if(null == oldparserDeviceHostDto)
{
log.info("设备{}不存在",imei);
return;
return null;
}
//解析和清洗body
... ... @@ -75,6 +75,7 @@ public abstract class BaseCallback<T> {
deviceCommandListenService.replySendMessage(oldparserDeviceHostDto.getId(),protocolPurificationModel.getNoticeMessage());
}
return protocolPurificationModel;
}
... ...
... ... @@ -8,6 +8,7 @@ import com.zhonglai.luhui.device.analysis.comm.dto.LogDeviceOperation;
import com.zhonglai.luhui.device.analysis.comm.dto.thingsmodels.ThingsModelDataTypeEnum;
import com.zhonglai.luhui.device.analysis.comm.dto.thingsmodels.ThingsModelItemBase;
import com.zhonglai.luhui.device.analysis.comm.dto.thingsmodels.specs.StringModelOutput;
import com.zhonglai.luhui.device.analysis.comm.factory.Topic;
import com.zhonglai.luhui.device.domain.IotThingsModel;
import com.zhonglai.luhui.device.protocol.factory.config.DeviceCach;
import com.zhonglai.luhui.device.protocol.factory.dto.ParserDeviceHostDto;
... ... @@ -81,6 +82,10 @@ public class DeviceCashUpService {
oldparserDeviceHostDto.set_data_up(true);
}
}
if(protocolPurificationModel.isAllUp())
{
oldparserDeviceHostDto.setData(newData);
}
}
JsonObject oldConfig = oldparserDeviceHostDto.getConfig();
... ... @@ -99,12 +104,17 @@ public class DeviceCashUpService {
oldparserDeviceHostDto.set_config_up(true);
}
}
if(protocolPurificationModel.isAllUp())
{
oldparserDeviceHostDto.setConfig(newConfig);
}
}
JsonObject oldSummary = oldparserDeviceHostDto.getSummary();
JsonObject newSummary = parserDeviceHostDto.getSummary();
if(null != newSummary)
{
if(null == oldSummary)
{
oldparserDeviceHostDto.setSummary(newSummary);
... ... @@ -117,11 +127,17 @@ public class DeviceCashUpService {
oldparserDeviceHostDto.set_summary_up(true);
}
}
if(protocolPurificationModel.isAllUp())
{
oldparserDeviceHostDto.setSummary(newSummary);
}
}
oldparserDeviceHostDto.setUpdateTime(parserDeviceHostDto.getUpdateTime());
oldparserDeviceHostDto.setOnline(parserDeviceHostDto.getOnline());
oldparserDeviceHostDto.setRssi(parserDeviceHostDto.getRssi());
oldparserDeviceHostDto.setSummary(parserDeviceHostDto.getSummary());
protocolPurificationModel.setParserDeviceHostDto(oldparserDeviceHostDto);
//更新主机缓存
... ... @@ -188,6 +204,10 @@ public class DeviceCashUpService {
oldparserDeviceInfoDto.set_data_up(true);
}
}
if(protocolPurificationModel.isAllUp())
{
oldparserDeviceInfoDto.setData(newData);
}
}
JsonObject oldConfig = oldparserDeviceInfoDto.getConfig();
... ... @@ -206,6 +226,10 @@ public class DeviceCashUpService {
oldparserDeviceInfoDto.set_conf_up(true);
}
}
if(protocolPurificationModel.isAllUp())
{
oldparserDeviceInfoDto.setConfig(newConfig);
}
}
return i;
}
... ...
... ... @@ -9,11 +9,16 @@ import com.zhonglai.luhui.device.analysis.comm.dto.DeviceSensorData;
import com.zhonglai.luhui.device.analysis.comm.dto.LogDeviceOperation;
import com.zhonglai.luhui.device.analysis.comm.dto.TableGenerateSqlEnum;
import com.zhonglai.luhui.device.domain.*;
import com.zhonglai.luhui.device.protocol.factory.control.DeviceCommandListenService;
import com.zhonglai.luhui.device.protocol.factory.dto.ParserDeviceHostDto;
import com.zhonglai.luhui.device.protocol.factory.dto.ParserDeviceInfoDto;
import com.zhonglai.luhui.device.protocol.factory.dto.ProtocolSubTopics;
import com.zhonglai.luhui.device.protocol.factory.service.PersistenceDBService;
import org.apache.rocketmq.spring.support.DefaultRocketMQListenerContainer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;
import java.util.ArrayList;
... ... @@ -26,6 +31,8 @@ public class DefaultDbService {
private BaseDao baseDao = new BaseDao();
@Autowired
private DeviceCommandListenService deviceCommandListenService;
public List<String> getTopicFromRole()
{
... ... @@ -162,8 +169,11 @@ public class DefaultDbService {
baseDao.updateBySql("UPDATE `iot_terminal` SET online=4 WHERE device_id=?",iotDevice.getClient_id());
break;
}
}else {
iotDevice.setStatus(3);
}
iotDevice.setOperation_token(deviceCommandListenService.getSelectorExpression());
int i = baseDao.update(iotDevice);
return i;
}
... ... @@ -188,6 +198,7 @@ public class DefaultDbService {
iotTerminal.setThings_model_config(parserDeviceInfoDto.getConfig().toString());
parserDeviceInfoDto.set_conf_up(false);
}
iotTerminal.setOperation_token(deviceCommandListenService.getSelectorExpression());
i+=baseDao.update(iotTerminal);
}
... ...
package com.zhonglai.luhui.http.service.proxy.controller;
import com.alibaba.fastjson.JSON;
import com.google.gson.JsonElement;
import com.google.gson.JsonObject;
import com.ruoyi.common.core.domain.AjaxResult;
import com.ruoyi.common.utils.GsonConstructor;
... ... @@ -11,6 +13,7 @@ import com.zhonglai.luhui.device.protocol.factory.comm.DataLogType;
import com.zhonglai.luhui.device.protocol.factory.comm.DeviceDataLog;
import com.zhonglai.luhui.device.protocol.factory.config.HttpNoticeCach;
import com.zhonglai.luhui.device.protocol.factory.dto.ParserDeviceHostDto;
import com.zhonglai.luhui.device.protocol.factory.dto.ParserDeviceInfoDto;
import com.zhonglai.luhui.device.protocol.factory.dto.ProtocolPurificationModel;
import com.zhonglai.luhui.device.protocol.factory.service.DeviceCashUpService;
import com.zhonglai.luhui.device.protocol.factory.service.PersistenceDBService;
... ... @@ -29,6 +32,7 @@ import org.springframework.web.bind.annotation.*;
import javax.servlet.http.HttpServletRequest;
import java.io.IOException;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.LinkedBlockingQueue;
... ... @@ -39,6 +43,7 @@ import java.util.concurrent.TimeUnit;
@RestController
@RequestMapping("/httpDataProxy")
public class HttpDataProxyController {
private static Logger log = LoggerFactory.getLogger(HttpDataProxyController.class);
@Autowired
private HttpCallback httpCallback;
... ... @@ -49,17 +54,66 @@ public class HttpDataProxyController {
@ApiImplicitParam(value = "设备imei",name = "imei"),
@ApiImplicitParam(value = "产品id",name = "role"),
@ApiImplicitParam(value = "用户名",name = "username"),
@ApiImplicitParam(value = "目前支持，ALL_POST 全量上报数据、ADD_POST 增量上报数据",name = "topicType"),
})
@RequestMapping(value = "post/{role}/{username}/{imei}/{time}",method = RequestMethod.POST)
public AjaxResult post(@PathVariable String role, @PathVariable String username, @PathVariable String imei, @PathVariable Integer time, HttpServletRequest request) throws IOException {
@RequestMapping(value = "post/{role}/{username}/{imei}/{time}/{topicType}",method = RequestMethod.POST)
public AjaxResult post(@PathVariable String role, @PathVariable String username, @PathVariable String imei, @PathVariable Integer time, @PathVariable String topicType, HttpServletRequest request) throws IOException {
byte[] bodyBytes = StreamUtils.copyToByteArray(request.getInputStream());
String body = new String(bodyBytes, request.getCharacterEncoding());
JsonObject jsonObject = GsonConstructor.get().fromJson(body,JsonObject.class);
String topic = "/"+role+"/"+username+"/"+imei+"/Json/ALL_POST/"+time;
String topic = "/"+role+"/"+username+"/"+imei+"/Json/"+topicType+"/"+time;
DeviceDataLog.info(imei, DataLogType.发来数据,topic,jsonObject);
httpCallback.messageArrived(imei,topic,jsonObject);
ProtocolPurificationModel protocolPurificationModel = httpCallback.messageArrived(imei,topic,jsonObject);
return AjaxResult.success(HttpNoticeCach.consumer(imei));
Map<String,Object> map = HttpNoticeCach.consumer(imei);
if (null == map)
{
Map<String,Map<String,Object>> rmap = new HashMap<>();
ParserDeviceHostDto parserDeviceHostDto = protocolPurificationModel.getParserDeviceHostDto();
if (null != parserDeviceHostDto.getConfig() && parserDeviceHostDto.getConfig().size()>0)
{
Map<String,Object> config= new HashMap<>();
for (String key:parserDeviceHostDto.getConfig().keySet())
{
JsonElement jsonElement = parserDeviceHostDto.getConfig().get(key);
if(jsonElement.isJsonObject())
{
JsonObject value = jsonElement.getAsJsonObject();
config.put(value.get("id").getAsString(),JSON.parse(value.get("value").toString()));
}
}
rmap.put("0",config);
}
List<ParserDeviceInfoDto> parserDeviceInfoDtoList = protocolPurificationModel.getParserDeviceInfoDtoList();
if (null != parserDeviceInfoDtoList && parserDeviceInfoDtoList.size() != 0)
{
for (ParserDeviceInfoDto parserDeviceInfoDto:parserDeviceInfoDtoList)
{
if (null != parserDeviceInfoDto.getConfig() && parserDeviceInfoDto.getConfig().size()>0)
{
Map<String,Object> config= new HashMap<>();
for (String key:parserDeviceInfoDto.getConfig().keySet())
{
JsonElement jsonElement = parserDeviceInfoDto.getConfig().get(key);
if(jsonElement.isJsonObject())
{
JsonObject value = jsonElement.getAsJsonObject();
config.put(value.get("id").getAsString(), JSON.parse(value.get("value").toString()));
}
}
rmap.put(parserDeviceInfoDto.getId().substring(parserDeviceInfoDto.getId().indexOf("_")+1),config);
}
}
}
if(rmap.size()!= 0)
{
log.info("返回的数据【{}】",rmap);
return AjaxResult.success(rmap);
}
}
log.info("返回的数据【{}】",map);
return AjaxResult.success(map);
}
}
... ...
... ... @@ -21,22 +21,13 @@ import java.util.TimerTask;
@Service
public class HttpCallback extends BaseCallback<JsonObject> {
@Autowired
private AsyncManager asyncManager;
public void messageArrived(String imei, String topic, JsonObject data)
public ProtocolPurificationModel messageArrived(String imei, String topic, JsonObject data)
{
try {
HttpCallback.super.messageArrived(imei,topic,data);
return HttpCallback.super.messageArrived(imei,topic,data);
} catch (Exception e) {
log.error(imei+"解析数据失败",e);
}
// asyncManager.execute(new TimerTask() {
// @Override
// public void run() {
//
// }
// });
return null;
}
}
... ...
... ... @@ -25,4 +25,5 @@ mqtt:
#rocketmq配置信息
rocketmq:
#nameservice服务器地址（多个以英文逗号隔开）
name-server: 47.115.144.179:9876
\ No newline at end of file
name-server: 47.115.144.179:9876
operationToken: ${random.uuid}
\ No newline at end of file
... ...
... ... @@ -28,9 +28,10 @@ mqtt:
client:
#客户端操作时间
operationTime: 10
productids: 27,23
# productids: 27,23
#rocketmq配置信息
rocketmq:
#nameservice服务器地址（多个以英文逗号隔开）
name-server: 47.115.144.179:9876
\ No newline at end of file
name-server: 47.115.144.179:9876
operationToken: ${random.uuid}
\ No newline at end of file
... ...
... ... @@ -591,6 +591,12 @@
<artifactId>canal.protocol</artifactId>
<version>${canal.version}</version>
</dependency>
<dependency>
<groupId>com.zhonglai</groupId>
<artifactId>weixin-api</artifactId>
<version>1.3.2</version>
</dependency>
</dependencies>
... ...