wisdomisite-java/src/main/java/com/zhgd/xmgl/async/AsyncCheckingPoint.java

package com.zhgd.xmgl.async;

import cn.hutool.core.collection.CollUtil;
import cn.hutool.core.date.DateUtil;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.toolkit.Wrappers;
import com.zhgd.mqtt.server.IMqttSender;
import com.zhgd.xmgl.modules.basicdata.entity.Notice;
import com.zhgd.xmgl.modules.basicdata.entity.SystemUser;
import com.zhgd.xmgl.modules.basicdata.mapper.NoticeMapper;
import com.zhgd.xmgl.modules.basicdata.mapper.SystemUserMapper;
import com.zhgd.xmgl.util.DateUtils;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Lazy;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Component;

import java.util.Collection;
import java.util.Date;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;

/**
 * @program: wisdomSite
 * @description: 异步AI分析数据处理
 * @author: Mr.Peng
 * @create: 2022-04-20 14:09
 **/
@Slf4j
@Component
public class AsyncCheckingPoint {
    @Lazy
    @Autowired
    private IMqttSender mqttPushClient;
    @Lazy
    @Autowired
    private SystemUserMapper systemUserMapper;
    @Lazy
    @Autowired
    private AsyncCommon asyncCommon;
    @Lazy
    @Autowired
    private NoticeMapper noticeMapper;

    /**
     * 通过用户id的集合发送信息到相关用户下的MQ以及app
     *
     * @param title
     * @param msg
     * @param itemType
     * @param userIdSet
     * @param payload
     */
    @Async("asyncExecutor")
    public synchronized void sendMqAndAppUserIdColl(String title, String msg, String itemType, Collection<String> userIdSet, String payload) {
        try {
            //向项目管理员和子账号推送通知
            List<SystemUser> systemUserList = systemUserMapper.selectList(Wrappers.lambdaQuery(SystemUser.class).in(SystemUser::getUserId, userIdSet));
            if (CollUtil.isNotEmpty(systemUserList)) {
                asyncCommon.sendMq(title, msg, itemType, systemUserList);
                this.addNoticeIfNotExist(title, msg, systemUserList);
                Map<String, List<SystemUser>> snUserMap = systemUserList.stream().filter(user -> user.getClientId() != null).collect(Collectors.groupingBy(SystemUser::getSn));
                for (Map.Entry<String, List<SystemUser>> snUserListEntry : snUserMap.entrySet()) {
                    asyncCommon.sendAppNotice(snUserListEntry.getKey(), title, msg, snUserListEntry.getValue(), payload);
                }
            }
        } catch (Exception e) {
            log.error("error：", e);
        }
    }

    private void addNoticeIfNotExist(String title, String msg, List<SystemUser> systemUserList) {
        for (SystemUser systemUser : systemUserList) {
            List<Notice> notices = noticeMapper.selectList(new LambdaQueryWrapper<Notice>().eq(Notice::getAccountId, systemUser.getUserId())
                    .ge(Notice::getSendTime, DateUtil.today())
                    .eq(Notice::getType, "11"));
            if (notices.stream().anyMatch(notice -> title.equals(notice.getTitle()) && msg.equals(notice.getMsg()))) {
                break;
            }
            Notice notice = new Notice();
            notice.setType("11");
            notice.setMsg(msg);
            notice.setAccountId(systemUser.getUserId());
            notice.setTitle(title);
            notice.setSendTime(DateUtil.formatDateTime(new Date()));
            noticeMapper.insert(notice);
        }
    }
}