message_bus / message_bus_service · 提交

转到一个项目

作者 朱兆平
提交 c3749b974102732927928211eba7c22a886e7122 1 个父辈 841e973e

批量新增路由与加载已配置路由适配

内嵌 并排对比
正在显示 9 个修改的文件 包含 110 行增加 和 53 行删除
... ... @@ -6,10 +6,7 @@ import com.sunyo.wlpt.message.bus.service.response.ResultJson;
import com.sunyo.wlpt.message.bus.service.service.RouterService;
import io.swagger.annotations.ApiOperation;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestBody;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;
import org.springframework.web.bind.annotation.*;
import java.util.List;
... ... @@ -27,4 +24,11 @@ public class RouterController {
boolean result = routerService.batchAddRouter(messageRouter);
return result ? new ResultJson("200","success") :new ResultJson("400","error");
}
@GetMapping("searchBysndr")
public ResultJson<List<MessageRouter>> getAllRouter(@RequestParam(value = "sndr",required = true) String sndr){
List<MessageRouter> all = routerService.searchBySndr(sndr);
return new ResultJson<>("200","success",all);
}
}
... ...
... ... @@ -2,6 +2,8 @@ package com.sunyo.wlpt.message.bus.service.mapper;
import com.sunyo.wlpt.message.bus.service.model.MessageRouter;
import java.util.List;
public interface MessageRouterMapper {
int deleteByPrimaryKey(String id);
... ... @@ -11,6 +13,8 @@ public interface MessageRouterMapper {
MessageRouter selectByPrimaryKey(String id);
List<MessageRouter> selectBySndr(String sndr);
int updateByPrimaryKeySelective(MessageRouter record);
int updateByPrimaryKey(MessageRouter record);
... ...
... ... @@ -13,6 +13,8 @@ public interface MessageTypeMapper {
MessageType selectByPrimaryKey(String id);
MessageType selectParentByChildName(String id);
List<MessageType> selectByParentId(String id);
List<MessageType> selectAll(MessageType record);
... ...
... ... @@ -16,6 +16,8 @@ public class MessageType extends BasePage{
private Integer type;
private boolean disabled;
private List<MessageType> children;
public String getId() {
... ... @@ -56,6 +58,9 @@ public class MessageType extends BasePage{
public void setType(Integer type) {
this.type = type;
if (type ==1 || type ==2 ){
this.disabled = true;
}
}
public List<MessageType> getChildren() {
... ... @@ -65,4 +70,12 @@ public class MessageType extends BasePage{
public void setChildren(List<MessageType> children) {
this.children = children;
}
public boolean isDisabled() {
return disabled;
}
public void setDisabled(boolean disabled) {
this.disabled = disabled;
}
}
\ No newline at end of file
... ...
... ... @@ -2,7 +2,11 @@ package com.sunyo.wlpt.message.bus.service.service;
import com.sunyo.wlpt.message.bus.service.model.MessageRouter;
import java.util.List;
public interface RouterService {
boolean batchAddRouter(MessageRouter messageRouter);
List<MessageRouter> searchBySndr(String sndr);
}
... ...
package com.sunyo.wlpt.message.bus.service.service.impl;
import com.sunyo.wlpt.message.bus.service.domain.BusQueue;
import com.sunyo.wlpt.message.bus.service.mapper.BusQueueMapper;
import com.sunyo.wlpt.message.bus.service.mapper.MessageRouterMapper;
import com.sunyo.wlpt.message.bus.service.mapper.MessageRouterReciverMapper;
import com.sunyo.wlpt.message.bus.service.mapper.UserTopicMapper;
import com.sunyo.wlpt.message.bus.service.mapper.*;
import com.sunyo.wlpt.message.bus.service.model.MessageRouter;
import com.sunyo.wlpt.message.bus.service.model.MessageRouterReciver;
import com.sunyo.wlpt.message.bus.service.model.MessageType;
... ... @@ -29,6 +26,9 @@ public class RouterServiceImp implements RouterService {
@Resource
BusQueueMapper busQueueMapper;
@Resource
MessageTypeMapper typeMapper;
@Transactional(rollbackFor = Exception.class)
@Override
... ... @@ -41,9 +41,9 @@ public class RouterServiceImp implements RouterService {
* 批量增加发送者表
*/
for (String sndr: senders) {
String id = IdUtils.generateId();
MessageRouter router = new MessageRouter();
router.setId(id);
router.setSndr(sndr);
router.setDes(record.getDes());
router.setVer(record.getVer());
... ... @@ -57,51 +57,55 @@ public class RouterServiceImp implements RouterService {
if (messageTypes!=null && !messageTypes.isEmpty()){
for (MessageType type: messageTypes) {
/**
* 判断消息类型是 大类 主类还是子类
* 只取二级类
*/
if (type.getType() == 2){
//大类处理
router.setBtype(type.getName());
}
if (type.getType() == 3){
//二级类处理
router.setStype(type.getName());
}
}
}
String id = IdUtils.generateId();
router.setId(id);
/**
* 查找二级类的上级类
*/
MessageType bType = typeMapper.selectParentByChildName(type.getName());
/**
* 路由生产者消息入库
*/
messageRouterMapper.insertSelective(router);
router.setBtype(bType.getName());
/**
* 批量增加接收者表
* 这里得receiver列表指的是接收者账号
* 要通过账号找到 账号所对应的队列信息 再入库
*/
List<String> receivers = record.getRcvrs();
if (receivers!=null && !receivers.isEmpty()){
for (String rcvr :receivers) {
String rcvr_id = IdUtils.generateId();
MessageRouterReciver messageRouterReciver = new MessageRouterReciver();
messageRouterReciver.setId(rcvr_id);
messageRouterReciver.setMessageRouterId(id);
/**
* 寻找账号对应的队列名称
*/
List<BusQueue> busQueues = busQueueMapper.selectTopicByUsername(rcvr);
messageRouterReciver.setRcvrTopic(busQueues.get(0).getQueueName());
/**
* 插入数据到订阅者表
*/
routerReciverMapper.insertSelective(messageRouterReciver);
//二级类处理
router.setStype(type.getName());
/**
* 路由生产者消息入库
*/
messageRouterMapper.insertSelective(router);
/**
* 批量增加接收者表
* 这里得receiver列表指的是接收者账号
* 要通过账号找到 账号所对应的队列信息 再入库
*/
List<String> receivers = record.getRcvrs();
if (receivers!=null && !receivers.isEmpty()){
for (String rcvr :receivers) {
String rcvr_id = IdUtils.generateId();
MessageRouterReciver messageRouterReciver = new MessageRouterReciver();
messageRouterReciver.setId(rcvr_id);
messageRouterReciver.setMessageRouterId(id);
/**
* 寻找账号对应的队列名称
*/
List<BusQueue> busQueues = busQueueMapper.selectTopicByUsername(rcvr);
messageRouterReciver.setRcvrTopic(busQueues.get(0).getQueueName());
/**
* 插入数据到订阅者表
*/
routerReciverMapper.insertSelective(messageRouterReciver);
}
}
}
}
}
}
}
return true;
... ... @@ -109,4 +113,14 @@ public class RouterServiceImp implements RouterService {
return false;
}
}
@Override
public List<MessageRouter> searchBySndr(String sndr) {
List<MessageRouter> list = messageRouterMapper.selectBySndr(sndr);
for (MessageRouter router:list) {
List<MessageRouterReciver> routerRecivers = routerReciverMapper.selectByRouterKey(router.getId());
router.setRcvrList(routerRecivers);
}
return list;
}
}
... ...
... ... @@ -179,10 +179,10 @@
#{gmtModified,jdbcType=TIMESTAMP},
</if>
<if test="partitionCount >0">
#{partitionCount,jdbcType=TIMESTAMP},
#{partitionCount,jdbcType=INTEGER},
</if>
<if test="queueType >0">
#{partitionCount,jdbcType=TIMESTAMP},
#{queueType,jdbcType=INTEGER},
</if>
</trim>
</insert>
... ...
... ... @@ -15,7 +15,6 @@
<result column="ver" property="ver" jdbcType="VARCHAR" />
<result column="creat_time" property="creatTime" jdbcType="TIMESTAMP" />
<result column="update_time" property="updateTime" jdbcType="TIMESTAMP" />
<collection column="id" javaType="java.util.ArrayList" ofType="com.sunyo.wlpt.message.bus.service.model.MessageRouterReciver" property="rcvrList" select="com.sunyo.wlpt.message.bus.service.mapper.selectByRouterKey" />
</resultMap>
<sql id="Base_Column_List" >
id, sndr, btype, stype, optype, msg_limit, `character`, status, `usage`, des, ver, creat_time,
... ... @@ -27,14 +26,22 @@
from message_router
where id = #{id,jdbcType=VARCHAR}
</select>
<select id="selectBySndr" resultMap="BaseResultMap" parameterType="java.lang.String" >
select
<include refid="Base_Column_List" />
from message_router
where sndr = #{id,jdbcType=VARCHAR}
</select>
<delete id="deleteByPrimaryKey" parameterType="java.lang.String" >
delete from message_router
where id = #{id,jdbcType=VARCHAR}
where id = #{sndr,jdbcType=VARCHAR}
</delete>
<insert id="insert" parameterType="com.sunyo.wlpt.message.bus.service.model.MessageRouter" >
insert into message_router (id, sndr, btype,
stype, optype, msg_limit,
`character`, status, `usage`, des,
`character`, `status`, `usage`, des,
ver, creat_time, update_time
)
values (#{id,jdbcType=VARCHAR}, #{sndr,jdbcType=VARCHAR}, #{btype,jdbcType=VARCHAR},
... ... @@ -68,7 +75,7 @@
`character`,
</if>
<if test="status != null" >
status,
`status`,
</if>
<if test="usage != null" >
`usage`,
... ...
... ... @@ -24,6 +24,15 @@
from message_type
where parent_id = #{parentId,jdbcType=VARCHAR}
</select>
<select id="selectParentByChildName" resultMap="BaseResultMap" parameterType="java.lang.String" >
select
<include refid="Base_Column_List" />
from message_type
where id =
(
SELECT parent_id from message_type where name = #{name,jdbcType=VARCHAR}
)
</select>
<select id="selectAll" parameterType="com.sunyo.wlpt.message.bus.service.model.MessageType" resultMap="BaseResultMap">
select
<include refid="Base_Column_List" />
... ...