feat(sse): sse验证及mapstruct

pom.xml

@@ -15,6 +15,7 @@
     <description>Portal Service</description>
     <properties>
         <java.version>1.8</java.version>
+        <mapstruct.version>1.5.2.Final</mapstruct.version>
     </properties>
     <dependencies>
         <!--tomcat-->
@@ -116,6 +117,23 @@
             <version>4.4</version>
         </dependency>
 
+        <dependency>
+            <groupId>com.google.guava</groupId>
+            <artifactId>guava</artifactId>
+            <version>23.0</version>
+        </dependency>
+
+        <dependency>
+            <groupId>org.mapstruct</groupId>
+            <artifactId>mapstruct</artifactId>
+            <version>${mapstruct.version}</version>
+        </dependency>
+
+        <dependency>
+            <groupId>org.mapstruct</groupId>
+            <artifactId>mapstruct-processor</artifactId>
+            <version>${mapstruct.version}</version>
+        </dependency>
     </dependencies>
 
     <build>
@@ -124,6 +142,36 @@
                 <groupId>org.springframework.boot</groupId>
                 <artifactId>spring-boot-maven-plugin</artifactId>
             </plugin>
+
+            <plugin>
+                <groupId>org.apache.maven.plugins</groupId>
+                <artifactId>maven-compiler-plugin</artifactId>
+                <configuration>
+                    <source>${java.version}</source>
+                    <target>${java.version}</target>
+                    <encoding>UTF-8</encoding>
+                    <annotationProcessorPaths>
+                        <!-- 此处，lombok依赖一定要放在，Mapstruct-processor依赖之前。否则，生成了maptruct的实现类，但该类只创建了对象，没有进行赋值 -->
+                        <path>
+                            <groupId>org.projectlombok</groupId>
+                            <artifactId>lombok</artifactId>
+                            <version>${lombok.version}</version>
+                        </path>
+                        <path>
+                            <groupId>org.projectlombok</groupId>
+                            <artifactId>lombok-mapstruct-binding</artifactId>
+                            <!-- 如果是0.1.0 有可能出现生成了MapStruct的实现类，但该类只创建了对象，没有进行赋值 -->
+                            <version>0.2.0</version>
+                        </path>
+                        <path>
+                            <groupId>org.mapstruct</groupId>
+                            <artifactId>mapstruct-processor</artifactId>
+                            <version>${mapstruct.version}</version>
+                        </path>
+                    </annotationProcessorPaths>
+                </configuration>
+            </plugin>
+
         </plugins>
     </build>
 

src/main/java/com/dragon/tj/portal/common/constants/BusinessConstants.java

@@ -0,0 +1,23 @@
+package com.dragon.tj.portal.common.constants;
+
+/**
+ * @author huey China.
+ * @Description : 业务常量
+ * @Date Created in 2023/6/16 11:13
+ */
+public interface BusinessConstants {
+
+
+    /**
+     * 1 通知公告
+     */
+    public static Integer DICT_ITEM_ID_1 = 1;
+    /**
+     * 2 信息交流
+     */
+    public static Integer DICT_ITEM_ID_2 = 2;
+    /**
+     * 3 标准规范
+     */
+    public static Integer DICT_ITEM_ID_3 = 3;
+}

src/main/java/com/dragon/tj/portal/common/constants/CommonConstants.java

@@ -37,5 +37,7 @@ public interface CommonConstants {
      */
     Integer FAIL = 1;
 
+    String sysCode = "sys_portal";
+
 
 }

src/main/java/com/dragon/tj/portal/common/convert/MessageReqToItemConvert.java

@@ -0,0 +1,21 @@
+package com.dragon.tj.portal.common.convert;
+
+
+import com.dragon.tj.portal.common.dto.message.MessageInfoItem;
+import com.dragon.tj.portal.common.dto.message.MessageInfoReq;
+import org.mapstruct.Mapper;
+import org.mapstruct.MappingConstants;
+
+@Mapper(componentModel = MappingConstants.ComponentModel.SPRING)
+public interface MessageReqToItemConvert {
+
+    /**
+     * User转换UserDTO
+     *
+     * @return {@link } Mappings是对字段名不一致的字段进行映射
+     * <p>
+     * Mapping可以处理变量名不一致映射关系以及指定日期格式等等
+     */
+    MessageInfoItem reqToItem(MessageInfoReq messageInfoReq);
+
+}

src/main/java/com/dragon/tj/portal/common/dto/message/MessageInfoItem.java

@@ -0,0 +1,68 @@
+package com.dragon.tj.portal.common.dto.message;
+
+
+import lombok.Getter;
+import lombok.Setter;
+import org.hibernate.validator.constraints.Range;
+
+import java.util.Set;
+
+/**
+ * @author huey China.
+ * @Description :
+ * @Date Created in 2023/6/15 15:59
+ */
+@Getter
+@Setter
+public class MessageInfoItem {
+
+
+    @Range(min = 1, max = 3, message = "请填入1~3,1 通知公告 2信息交流 3 标准规范 ")
+    private Integer messageType;
+
+    /**
+     * 范围ids
+     */
+    private Set<String> scopeIds;
+
+    /**
+     * 阅读状态 默认0未读 1已读
+     */
+    private Integer readStatus;
+
+    /**
+     * 信息标题
+     */
+    private String title;
+
+    /**
+     * 信息内容
+     */
+    private String content;
+
+    /**
+     * 附件地址
+     */
+    private String attachmentPath;
+
+
+    /**
+     * 版本号
+     */
+    private Integer version;
+
+    /**
+     *
+     *--------------非req参数
+     */
+    /**
+     * TODO appId
+     */
+    private String appId = "appId";
+
+    /**
+     * 客户端id
+     */
+    private String clientId;
+
+}

src/main/java/com/dragon/tj/portal/common/dto/MessageInfoDTO.java → src/main/java/com/dragon/tj/portal/common/dto/message/MessageInfoReq.java

@@ -1,10 +1,10 @@
-package com.dragon.tj.portal.common.dto;
+package com.dragon.tj.portal.common.dto.message;
 
 
 import lombok.Getter;
 import lombok.Setter;
+import org.hibernate.validator.constraints.Range;
 
-import javax.validation.constraints.NotBlank;
 import java.util.Set;
 
 /**
@@ -14,23 +14,16 @@ import java.util.Set;
  */
 @Getter
 @Setter
-public class MessageInfoDTO {
+public class MessageInfoReq {
 
 
-    private Long id;
-
-    @NotBlank(message = "消息类型不能为空")
+    @Range(min = 1, max = 3, message = "请填入1~3,1 通知公告 2信息交流 3 标准规范 ")
     private Integer messageType;
 
     /**
-     * 部门ids
+     * 范围ids
      */
-    private Set<String> deptIds;
-
-    /**
-     * 联系人ids
-     */
-    private Set<String> contractIds;
+    private Set<String> scopeIds;
 
     /**
      * 阅读状态 默认0未读 1已读
@@ -58,4 +51,15 @@ public class MessageInfoDTO {
      */
     private Integer version;
 
+    /**
+     *
+     *--------------非req参数
+     */
+    /**
+     * TODO appId
+     */
+    private String appId = "appId";
+
+    private Set<String> messageClientIds;
+
 }

src/main/java/com/dragon/tj/portal/common/enums/MessageInfoErrorEnums.java

@@ -0,0 +1,49 @@
+/*
+ * All rights reserved.  http://www.sunac.com.cn
+ *
+ * This software is the confidential and proprietary information of Sunac Tech Corporation ("Confidential Information").
+ * You shall not disclose such Confidential Information and shall use
+ * it only in accordance with the terms of the license agreement
+ * you entered into with Sunac Tech.
+ */
+/**
+ *
+ */
+package com.dragon.tj.portal.common.enums;
+
+import com.dragon.tj.portal.component.exception.base.FrameResultError;
+import lombok.Getter;
+
+/**
+ * @author huey China.
+ * @Description : 模块异常-信息管理
+ * @Date Created in 2023/6/16 11:32
+ */
+public interface MessageInfoErrorEnums extends FrameResultError {
+
+
+    @Getter
+    enum Code implements FrameResultError {
+
+        MESSAGE_CONTRACTS_EMPTY(201001, "信息交流的联系人不能为空!"),
+        MESSAGE_DEPT_EMPTY(201002, "当前信息类型部门名称不能为空!"),
+
+
+        ;
+
+
+        Code(int code, String msg) {
+            this.code = code;
+            this.msg = msg;
+        }
+
+        /**
+         * 错误码.
+         */
+        private int code;
+        /**
+         * 描述
+         */
+        private String msg;
+    }
+}

src/main/java/com/dragon/tj/portal/component/exception/message/MessageInfoException.java

@@ -0,0 +1,21 @@
+package com.dragon.tj.portal.component.exception.message;
+
+import com.dragon.tj.portal.common.constants.CommonConstants;
+import com.dragon.tj.portal.component.exception.base.FrameAbstractException;
+import com.dragon.tj.portal.component.exception.base.ResultError;
+
+/**
+ * @author huey China.
+ * @Description : 信息相关异常
+ * @Date Created in 2023/6/16 11:30
+ */
+public class MessageInfoException extends FrameAbstractException {
+    public MessageInfoException(ResultError resultError) {
+        super(resultError);
+    }
+
+    @Override
+    public String setSystemCode() {
+        return CommonConstants.sysCode;
+    }
+}

src/main/java/com/dragon/tj/portal/component/message/Consumer.java → src/main/java/com/dragon/tj/portal/component/message/MessageConsumer.java

@@ -1,7 +1,11 @@
 package com.dragon.tj.portal.component.message;
 
 import cn.hutool.core.util.StrUtil;
+import com.alibaba.fastjson.JSON;
 import com.alibaba.fastjson.JSONObject;
+import com.dragon.tj.portal.common.convert.MessageReqToItemConvert;
+import com.dragon.tj.portal.common.dto.message.MessageInfoItem;
+import com.dragon.tj.portal.common.dto.message.MessageInfoReq;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.http.MediaType;
@@ -11,6 +15,7 @@ import org.springframework.stereotype.Component;
 
 import java.io.IOException;
 import java.util.Map;
+import java.util.Set;
 
 /**
  * @author huey China.
@@ -19,11 +24,14 @@ import java.util.Map;
  */
 @Component
 @Slf4j
-public class Consumer {
+public class MessageConsumer {
 
     @Autowired
     private Map<String, OpenApiSseEmitter> sseEmitters;
 
+    @Autowired
+    private MessageReqToItemConvert messageReqToItemConvert;
+
     /**
      * get client published message from kafka，process it according to the serverId
      *
@@ -33,34 +41,41 @@ public class Consumer {
 //    @KafkaListener(topics = {"sseTopic"}, clientIdPrefix = "${spring.application.name}", id = "openapi-be-sse-connection-", idIsGroup = false, groupId = "openapi-be-sse-connection-" + "#{T(java.util.UUID).randomUUID()}")
     @KafkaListener(topics = {"sseTopic"})
     public void sseConnectionProcess(String msg, Acknowledgment ack) {
-//        log.info("get kafka msg from topic:{}, msg:{}", customProperty.getKafka().getTopics().getSseConnection().getTopic(), msg);
         log.info("get kafka msg from topic:{}, msg:{}", KafkaInitialConfiguration.sseTopic, msg);
         if (StrUtil.isEmpty(msg)) {
             log.error("kafka msg is empty, no process");
             return;
         }
-        SsePublishReqDto reqDto;
+        MessageInfoReq reqDTO;
         try {
-            reqDto = JSONObject.parseObject(msg, SsePublishReqDto.class);
+            reqDTO = JSONObject.parseObject(msg, MessageInfoReq.class);
         } catch (Exception e) {
             log.error("parsing string to obj failed, msg={}, e={}", msg, e);
             return;
         }
 
-        OpenApiSseEmitter emitter = sseEmitters.get(reqDto.getClientId());
-        if (emitter == null) {
-            log.error("can't find the sseEmitter obj from sseEmitters, no process");
-            return;
-        }
+        Set<String> messageClientIds = reqDTO.getMessageClientIds();
 
-        try {
-            // 发送消息给客户端
-            log.info("send sse msg={} to clientId={}", reqDto.getMessage(), reqDto.getClientId());
-            emitter.send(reqDto.getMessage(), MediaType.APPLICATION_JSON);
-            ack.acknowledge();
-        } catch (IOException e) {
-            emitter.completeWithError(e);
+        for (String clientId : messageClientIds) {
+            OpenApiSseEmitter emitter = sseEmitters.get(clientId);
+
+            if (emitter == null) {
+                log.error("can't find the sseEmitter obj from sseEmitters, no process {}",clientId);
+                continue;
+            }
+            try {
+                MessageInfoItem messageInfoItem = messageReqToItemConvert.reqToItem(reqDTO);
+                messageInfoItem.setClientId(clientId);
+
+                String sendMsg = JSON.toJSONString(messageInfoItem);
+                // 发送消息给客户端
+                log.info("send sse msg={} to clientId={}", sendMsg, clientId);
+                emitter.send(sendMsg, MediaType.APPLICATION_JSON);
+            } catch (IOException e) {
+                emitter.completeWithError(e);
+            }
         }
+        ack.acknowledge();
     }
 
 }

src/main/java/com/dragon/tj/portal/component/message/Producer.java → src/main/java/com/dragon/tj/portal/component/message/MessageProducer.java

@@ -16,7 +16,7 @@ import javax.annotation.Resource;
 */
 @Component
 @Slf4j
-public class Producer {
+public class MessageProducer {
 
     @Autowired
     private KafkaTemplate<String, Object> kafkaTemplate;

src/main/java/com/dragon/tj/portal/component/message/SseController.java

@@ -2,7 +2,11 @@ package com.dragon.tj.portal.component.message;
 
 import com.alibaba.fastjson.JSON;
 import com.dragon.tj.portal.common.base.R;
+import com.dragon.tj.portal.common.dto.message.MessageInfoReq;
+import com.dragon.tj.portal.service.MessageInfoService;
 import lombok.extern.slf4j.Slf4j;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.validation.annotation.Validated;
 import org.springframework.web.bind.annotation.*;
 import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;
 
@@ -22,8 +26,8 @@ public class SseController {
     @Resource
     private Map<String, OpenApiSseEmitter> sseEmitters;
 
-    @Resource
-    private Producer producer;
+    @Autowired
+    private MessageInfoService messageInfoService;
 
     @GetMapping("/subscribe/{clientId}")
     public SseEmitter subscribe(@PathVariable String clientId, HttpServletResponse response) {
@@ -41,12 +45,10 @@ public class SseController {
         return emitter;
     }
 
-    @PostMapping("/publish/{clientId}")
-    public R publish(@PathVariable String clientId, @RequestBody String message) {
-
-        SsePublishReqDto build = SsePublishReqDto.builder().clientId(clientId).message(message).serverId("appId").build();
-        boolean send = producer.send(KafkaInitialConfiguration.sseTopic, JSON.toJSONString(build));
-        return R.ok(send);
+    @PostMapping("/publish")
+    public R publish(@Validated @RequestBody MessageInfoReq messageInfoReq) {
+        log.info("sseController req param is {}", JSON.toJSONString(messageInfoReq));
+        return messageInfoService.push(messageInfoReq);
     }
 
 

src/main/java/com/dragon/tj/portal/controller/MessageInfoController.java

@@ -16,11 +16,6 @@ import org.springframework.web.bind.annotation.RestController;
 @RequestMapping("/messageInfo")
 public class MessageInfoController {
 
-    /**
-     *
-     *发布
-     */
-
     /**
      *
      *详情

src/main/java/com/dragon/tj/portal/service/MessageInfoService.java

@@ -1,5 +1,7 @@
 package com.dragon.tj.portal.service;
 
+import com.dragon.tj.portal.common.base.R;
+import com.dragon.tj.portal.common.dto.message.MessageInfoReq;
 import com.dragon.tj.portal.entity.MessageInfo;
 import com.baomidou.mybatisplus.extension.service.IService;
 
@@ -13,4 +15,5 @@ import com.baomidou.mybatisplus.extension.service.IService;
  */
 public interface MessageInfoService extends IService<MessageInfo> {
 
+    R push(MessageInfoReq messag);
 }

src/main/java/com/dragon/tj/portal/service/impl/MessageInfoServiceImpl.java

@@ -1,11 +1,21 @@
 package com.dragon.tj.portal.service.impl;
 
+import com.alibaba.fastjson.JSON;
+import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
+import com.dragon.tj.portal.common.base.R;
+import com.dragon.tj.portal.common.constants.BusinessConstants;
+import com.dragon.tj.portal.common.dto.message.MessageInfoReq;
+import com.dragon.tj.portal.component.message.KafkaInitialConfiguration;
+import com.dragon.tj.portal.component.message.MessageProducer;
 import com.dragon.tj.portal.entity.MessageInfo;
 import com.dragon.tj.portal.mapper.MessageInfoMapper;
 import com.dragon.tj.portal.service.MessageInfoService;
-import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
+import com.google.common.collect.Sets;
+import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.stereotype.Service;
 
+import java.util.Set;
+
 /**
  * <p>
  * 信息表 服务实现类
@@ -17,4 +27,52 @@ import org.springframework.stereotype.Service;
 @Service
 public class MessageInfoServiceImpl extends ServiceImpl<MessageInfoMapper, MessageInfo> implements MessageInfoService {
 
+    @Autowired
+    private MessageProducer producer;
+
+    @Override
+    public R push(MessageInfoReq messageInfoReq) {
+        return R.ok(this.doMessage(messageInfoReq));
+    }
+
+    /**
+     * @author huey China.
+     * @Description : 预先处理 TODO message 没走字典
+     * @Date Created in 2023/6/16 11:25
+     */
+    private boolean doMessage(MessageInfoReq messageInfoReq) {
+
+        this.validate(messageInfoReq);
+        return this.transferIds(messageInfoReq);
+    }
+
+    /**
+     * @author huey China.
+     * @Description : TODO 跟进选中范围 填充不同信息类型的用户标识messageClientIds
+     * @Date Created in 2023/6/16 11:44
+     */
+    private boolean transferIds(MessageInfoReq messageInfoReq) {
+
+        boolean isSend = false;
+        Integer messageType = messageInfoReq.getMessageType();
+        Set<String> scopeIds = messageInfoReq.getScopeIds();
+
+        //测试部门1
+        if (messageType.equals(BusinessConstants.DICT_ITEM_ID_2) && scopeIds.contains("1")) {
+            //登录人标识
+            Set<String> messageClientIds = Sets.newHashSet();
+            messageClientIds.add("c1");
+            messageClientIds.add("c2");
+            messageInfoReq.setMessageClientIds(messageClientIds);
+
+            isSend = producer.send(KafkaInitialConfiguration.sseTopic, JSON.toJSONString(messageInfoReq));
+        }
+        return isSend;
+    }
+
+
+    private void validate(MessageInfoReq messageInfoReq) {
+
+    }
+
 }