|
|
|
@ -15,14 +15,20 @@ import com.ccsens.ccmq.lowlevel.service.IUserService; |
|
|
|
import com.fasterxml.jackson.databind.JsonNode; |
|
|
|
import org.slf4j.Logger; |
|
|
|
import org.slf4j.LoggerFactory; |
|
|
|
import org.springframework.amqp.core.AmqpTemplate; |
|
|
|
import org.springframework.beans.factory.annotation.Autowired; |
|
|
|
import org.springframework.scheduling.annotation.Async; |
|
|
|
import org.springframework.scheduling.annotation.Scheduled; |
|
|
|
import org.springframework.stereotype.Component; |
|
|
|
import wiki.tall.ccmq.common.bean.dto.ccmodbus.CCModBusEntity; |
|
|
|
import wiki.tall.ccmq.common.bean.dto.ccmodbus.Register; |
|
|
|
import wiki.tall.ccmq.common.config.SettingProps; |
|
|
|
import wiki.tall.ccmq.common.util.*; |
|
|
|
|
|
|
|
import javax.annotation.PostConstruct; |
|
|
|
import java.util.HashSet; |
|
|
|
import java.util.List; |
|
|
|
import java.util.Map; |
|
|
|
import java.util.Set; |
|
|
|
|
|
|
|
/** |
|
|
|
@ -40,6 +46,9 @@ public class MessageHandler { |
|
|
|
*/ |
|
|
|
private static final Integer MAX_MESSAGE_NUM = 50; |
|
|
|
|
|
|
|
@Autowired |
|
|
|
private AmqpTemplate rabbitTemplate; |
|
|
|
|
|
|
|
private static IMessageDao getMessageDao(){ |
|
|
|
return SpringContextUtils.getBean(IMessageDao.class); |
|
|
|
} |
|
|
|
@ -52,6 +61,13 @@ public class MessageHandler { |
|
|
|
return SpringContextUtils.getBean(IUserService.class); |
|
|
|
} |
|
|
|
|
|
|
|
private static MessageHandler messageHandler; |
|
|
|
@PostConstruct |
|
|
|
public void init(){ |
|
|
|
messageHandler = this; |
|
|
|
messageHandler.rabbitTemplate = this.rabbitTemplate; |
|
|
|
} |
|
|
|
|
|
|
|
/** |
|
|
|
* 每个20s同步一次有未决消息的用户到pendingClientSet中 |
|
|
|
* @throws Exception |
|
|
|
@ -240,6 +256,7 @@ public class MessageHandler { |
|
|
|
} |
|
|
|
|
|
|
|
private static void handlerServerMessage(String type, InMessage inMessage) throws Exception { |
|
|
|
logger.info("处理Server消息：{},{}", type, inMessage); |
|
|
|
MessageConstant.ClientMessageType clientMessageType = MessageConstant.ClientMessageType.valueOf(type); |
|
|
|
String data = inMessage.getData(); |
|
|
|
OutMessage outMessage = null; |
|
|
|
@ -255,16 +272,27 @@ public class MessageHandler { |
|
|
|
break; |
|
|
|
} |
|
|
|
case Auth: { |
|
|
|
logger.info("授权开始"); |
|
|
|
boolean authSuccess = false; |
|
|
|
AuthMessage inSysData = JacksonUtil.jsonToBean(data, AuthMessage.class); |
|
|
|
if(null != inSysData.getData()){ |
|
|
|
|
|
|
|
if(StrUtil.isNotEmpty(inSysData.getData().getToken())) { |
|
|
|
logger.info("token授权"); |
|
|
|
String userId = getUserService().getUserIdByToken(inSysData.getData().getToken()); |
|
|
|
if(StrUtil.isNotEmpty(userId)){ |
|
|
|
ChannelManager.authChannel(ChannelManager.getCurrentChannel(),userId,inSysData.getData().getMajor(),inSysData.getData().getMinor(), inSysData.getMessage()); |
|
|
|
onClientOnLine(MessageConstant.DomainType.User,userId); |
|
|
|
authSuccess = true; |
|
|
|
} |
|
|
|
} else if (inSysData.getData().getAuthId() != null) { |
|
|
|
logger.info("authid授权"); |
|
|
|
// authId有值默认授权通过
|
|
|
|
ChannelManager.authChannel(ChannelManager.getCurrentChannel(),String.valueOf(inSysData.getData().getAuthId()),inSysData.getData().getMajor(),inSysData.getData().getMinor(), inSysData.getMessage()); |
|
|
|
logger.info("授权完成"); |
|
|
|
onClientOnLine(MessageConstant.DomainType.User,String.valueOf(inSysData.getData().getAuthId())); |
|
|
|
logger.info("设置在线"); |
|
|
|
authSuccess = true; |
|
|
|
} |
|
|
|
} |
|
|
|
if(!authSuccess){ |
|
|
|
@ -272,10 +300,12 @@ public class MessageHandler { |
|
|
|
new ChannelStatusMessage(false,0L,MessageConstant.Error.AuthFailed)) |
|
|
|
); |
|
|
|
}else{ |
|
|
|
|
|
|
|
outMessage = new OutMessage(JacksonUtil.beanToJson( |
|
|
|
new ChannelStatusMessage(true,0L,MessageConstant.Error.Ok)) |
|
|
|
); |
|
|
|
} |
|
|
|
logger.info("授权结束"); |
|
|
|
break; |
|
|
|
} |
|
|
|
case GetChannelStatus: { |
|
|
|
@ -402,6 +432,68 @@ public class MessageHandler { |
|
|
|
) |
|
|
|
); |
|
|
|
break; |
|
|
|
} |
|
|
|
case ModBus:{ |
|
|
|
//1.按照功能码读写数据
|
|
|
|
ModBusMessage modBusMessage = JacksonUtil.jsonToBean(data, ModBusMessage.class); |
|
|
|
byte oper = modBusMessage.getData().getOper(); |
|
|
|
switch (oper) { |
|
|
|
case CCModBusEntity.Oper.READ : { |
|
|
|
// TODO 读操作
|
|
|
|
WrapperedChannel wrapperedChannel = ChannelManager.getWrapperedChannelByChannel(ChannelManager.getCurrentChannel()); |
|
|
|
Object o = wrapperedChannel.getRegister().get(modBusMessage.getData().getRegister()); |
|
|
|
break; |
|
|
|
} |
|
|
|
case CCModBusEntity.Oper.WRITE : { |
|
|
|
// TODO 写单个寄存器
|
|
|
|
break; |
|
|
|
} |
|
|
|
case CCModBusEntity.Oper.WRITE_MORE : { |
|
|
|
// 写多个寄存器
|
|
|
|
// 2. 如果是写操作且值发生改变，放到mq里
|
|
|
|
byte[] modBusData = modBusMessage.getData().getModBusData(); |
|
|
|
// 读起始寄存器地址
|
|
|
|
int startAddr = CCModBusEntity.getRegisterStartAddr(modBusData); |
|
|
|
int num = CCModBusEntity.getRegisterStartNum(modBusData); |
|
|
|
WrapperedChannel wrapperedChannel = ChannelManager.getWrapperedChannelByChannel(ChannelManager.getCurrentChannel()); |
|
|
|
Map<String, Long> register = wrapperedChannel.getRegister(); |
|
|
|
// 授权数据
|
|
|
|
Long authId = register.get(String.valueOf(MessageConstant.Register.Auth.getAddr())); |
|
|
|
for (int i = 0; i < num; i++) { |
|
|
|
// 寄存器的原始数据
|
|
|
|
Long origin = register.get(String.valueOf(startAddr)); |
|
|
|
// 寄存器将要修改的数据
|
|
|
|
long value = 0; |
|
|
|
for (int j = 0; j < CCModBusEntity.SIZE_DATA_SINGLE; j++) { |
|
|
|
value <<= 8; |
|
|
|
value += modBusData[CCModBusEntity.POSITION_DATA + CCModBusEntity.SIZE_DATA_SINGLE * i + j]; |
|
|
|
} |
|
|
|
if (origin == null || origin.longValue() != value) { |
|
|
|
register.put(String.valueOf(startAddr), value); |
|
|
|
logger.info("寄存器数据：{}", register); |
|
|
|
// 数据不一致，放到消息队列里
|
|
|
|
Register register1 = new Register(); |
|
|
|
register1.setAuthId(authId); |
|
|
|
register1.setAddr(startAddr+i); |
|
|
|
register1.setValue(value); |
|
|
|
OutMessage outMessage1 = new OutMessage(JacksonUtil.beanToJson(register1)); |
|
|
|
OutMessageSet set = new OutMessageSet(); |
|
|
|
set.add(outMessage1); |
|
|
|
messageHandler.rabbitTemplate.convertAndSend("wisdom_car", |
|
|
|
JacksonUtil.beanToJson(set)); |
|
|
|
logger.info("modbus通知mq:{}", set); |
|
|
|
|
|
|
|
} |
|
|
|
|
|
|
|
} |
|
|
|
// 3. 封装返回值
|
|
|
|
outMessage = new OutMessage(JacksonUtil.beanToJson(new ModBusAckMessage(modBusData))); |
|
|
|
break; |
|
|
|
} |
|
|
|
default:break; |
|
|
|
} |
|
|
|
|
|
|
|
|
|
|
|
} |
|
|
|
default: break; |
|
|
|
} |
|
|
|
@ -416,7 +508,9 @@ public class MessageHandler { |
|
|
|
|
|
|
|
private static void onClientOnLine(MessageConstant.DomainType domain,String userId){ |
|
|
|
getMessageDao().updateMessageStatusToPending(domain,userId); |
|
|
|
logger.info("修改mongodb"); |
|
|
|
addClientToRedisUnPendingSet(CcMessageUtil.getDomainTypeAndUserIdString(domain,userId)); |
|
|
|
logger.info("添加redisUnPending"); |
|
|
|
} |
|
|
|
|
|
|
|
private static void updateMessageAckStatus(String ackId) throws Exception { |
|
|
|
@ -477,16 +571,22 @@ public class MessageHandler { |
|
|
|
String []stringArray = CcMessageUtil.splitTypeAndUserId(domainTypeAndUserId); |
|
|
|
MessageConstant.DomainType toDomain = MessageConstant.DomainType.valueOf(stringArray[0]); |
|
|
|
String to = stringArray[1]; |
|
|
|
logger.info("查询ack"); |
|
|
|
//1.1 当前用户是否在pendingClients列表中
|
|
|
|
if (!RedisUtil.sHas(redisUnPendingClientSetKey,domainTypeAndUserId)) { |
|
|
|
logger.info("当前用户不在pendingClients列表中"); |
|
|
|
//1.2 当前用户是否在redis的等待ack列表中
|
|
|
|
if(!RedisUtil.hasKey(redisWaitAckSetKey)){ |
|
|
|
logger.info("当前用户不在redis的等待ack列表中"); |
|
|
|
//1.3 当前用户是否有pending消息
|
|
|
|
if(getMessageDao().countClientPendingMessage(toDomain,to) > 0){ |
|
|
|
logger.info("当前用户有pending消息"); |
|
|
|
//1.4 将当前用户添加到redisUnPendingClientSet列表中排队处理
|
|
|
|
RedisUtil.sSet(redisUnPendingClientSetKey,domainTypeAndUserId); |
|
|
|
logger.info("将当前用户添加到redisUnPendingClientSet列表中排队处理"); |
|
|
|
} |
|
|
|
} |
|
|
|
} |
|
|
|
logger.info("添加ack"); |
|
|
|
} |
|
|
|
} |
|
|
|
|
