This commit is contained in:
kerwincui
2024-03-17 14:59:23 +08:00
parent 3d44f4674c
commit 5539c1b6af
999 changed files with 115642 additions and 10757 deletions

View File

@@ -0,0 +1,12 @@
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<parent>
<artifactId>fastbee-server</artifactId>
<groupId>com.fastbee</groupId>
<version>3.8.5</version>
</parent>
<artifactId>base-server</artifactId>
</project>

View File

@@ -0,0 +1,29 @@
package com.fastbee.base.codec;
/**
* 分隔符报文处理器
* 处理分割符报文tcp粘包处理
* @author bill
*/
public class Delimiter {
public final byte[] value;
public final boolean strip;
public Delimiter(byte[] value) {
this(value, true);
}
public Delimiter(byte[] value, boolean strip) {
this.value = value;
this.strip = strip;
}
public byte[] getValue() {
return value;
}
public boolean isStrip() {
return strip;
}
}

View File

@@ -0,0 +1,76 @@
package com.fastbee.base.codec;
import static io.netty.util.internal.ObjectUtil.checkPositive;
import static io.netty.util.internal.ObjectUtil.checkPositiveOrZero;
/**
* 固定长度报文处理器
* 处理固定长度报文tcp粘包处理
* @author bill
*/
public class LengthField {
public final byte[] prefix;
/*最大帧长度*/
public final int lengthFieldMaxFrameLength;
/*偏移量*/
public final int lengthFieldOffset;
/*字段长度*/
public final int lengthFieldLength;
/*结尾偏移量*/
public final int lengthFieldEndOffset;
/*报文调整 默认0不调整*/
public final int lengthAdjustment;
/*默认0*/
public final int initialBytesToStrip;
/**构造固定长度处理器*/
public LengthField(byte[] prefix, int maxFrameLength, int lengthFieldOffset, int lengthFieldLength) {
this(prefix, maxFrameLength, lengthFieldOffset, lengthFieldLength, 0, 0);
}
public LengthField(byte[] prefix, int maxFrameLength, int lengthFieldOffset, int lengthFieldLength, int lengthAdjustment, int initialBytesToStrip) {
checkPositive(maxFrameLength, "maxFrameLength_LengthField");
checkPositiveOrZero(lengthFieldOffset, "lengthFieldOffset");
checkPositiveOrZero(initialBytesToStrip, "initialBytesToStrip");
if (lengthFieldOffset > maxFrameLength - lengthFieldLength) {
throw new IllegalArgumentException("maxFrameLength_LengthField (" + maxFrameLength + ") must be equal to or greater than lengthFieldOffset (" + lengthFieldOffset + ") + lengthFieldLength (" + lengthFieldLength + ").");
} else {
this.prefix = prefix;
this.lengthFieldMaxFrameLength = maxFrameLength;
this.lengthFieldOffset = lengthFieldOffset;
this.lengthFieldLength = lengthFieldLength;
this.lengthAdjustment = lengthAdjustment;
this.lengthFieldEndOffset = lengthFieldOffset + lengthFieldLength;
this.initialBytesToStrip = initialBytesToStrip;
}
}
public byte[] getPrefix() {
return prefix;
}
public int getLengthFieldMaxFrameLength() {
return lengthFieldMaxFrameLength;
}
public int getLengthFieldOffset() {
return lengthFieldOffset;
}
public int getLengthFieldLength() {
return lengthFieldLength;
}
public int getLengthFieldEndOffset() {
return lengthFieldEndOffset;
}
public int getLengthAdjustment() {
return lengthAdjustment;
}
public int getInitialBytesToStrip() {
return initialBytesToStrip;
}
}

View File

@@ -0,0 +1,18 @@
package com.fastbee.base.codec;
import com.fastbee.common.core.mq.DeviceReport;
import io.netty.buffer.ByteBuf;
/**
* 基础消息解码类
*
* @author bill
*/
public interface MessageDecoder {
/**
* TCP3.进站消息解码方法
*/
DeviceReport decode(ByteBuf buf, String clientId);
}

View File

@@ -0,0 +1,15 @@
package com.fastbee.base.codec;
import com.fastbee.common.core.protocol.Message;
import com.fastbee.base.session.Session;
import io.netty.buffer.ByteBuf;
/**
* 基础消息编码类
*
* @author bill
*/
public interface MessageEncoder{
ByteBuf encode(Message message, String clientId);
}

View File

@@ -0,0 +1,61 @@
package com.fastbee.base.core;
import com.fastbee.base.core.annotation.Async;
import com.fastbee.base.core.annotation.AsyncBatch;
import com.fastbee.base.core.annotation.PakMapping;
import com.fastbee.base.core.hanler.AsyncBatchHandler;
import com.fastbee.base.core.hanler.BaseHandler;
import com.fastbee.base.core.hanler.SyncHandler;
import java.lang.reflect.Method;
import java.util.HashMap;
import java.util.Map;
/**
* 消息处理映射
*
* @author bill
*/
public abstract class AbstractHandlerMapping implements HandlerMapping {
private final Map<Object, BaseHandler> handlerMap = new HashMap<>(64);
/**
* 将node中被@Column标记的方法注册到映射表
*/
protected synchronized void registerHandlers(Object bean) {
Class<?> beanClass = bean.getClass();
Method[] methods = beanClass.getDeclaredMethods();
for (Method method : methods) {
PakMapping annotation = method.getAnnotation(PakMapping.class);
if (annotation != null) {
String desc = annotation.desc();
int[] types = annotation.types();
AsyncBatch asyncBatch = method.getAnnotation(AsyncBatch.class);
BaseHandler baseHandler;
// 异步处理
if (asyncBatch != null) {
baseHandler = new AsyncBatchHandler(bean, method, desc, asyncBatch.poolSize(), asyncBatch.maxMessageSize(), asyncBatch.maxWaitTime());
} else {
baseHandler = new SyncHandler(bean, method, desc, method.isAnnotationPresent(Async.class));
}
for (int type : types) {
handlerMap.put(type, baseHandler);
}
}
}
}
/**
* 根据消息类型获取handler
*/
@Override
public BaseHandler getHandler(int messageId) {
return handlerMap.get(messageId);
}
}

View File

@@ -0,0 +1,27 @@
package com.fastbee.base.core;
import com.fastbee.base.core.annotation.Node;
import com.fastbee.base.util.ClassUtils;
import java.util.List;
/**
* 默认消息映射处理类
* @author bill
*/
public class DefaultHandlerMapping extends AbstractHandlerMapping {
public DefaultHandlerMapping(String endpointPackage) {
List<Class> endpointClasses = ClassUtils.getClassList(endpointPackage, Node.class);
for (Class endpointClass : endpointClasses) {
try {
Object bean = endpointClass.getDeclaredConstructor((Class[]) null).newInstance((Object[]) null);
super.registerHandlers(bean);
} catch (Exception e) {
throw new RuntimeException(e);
}
}
}
}

View File

@@ -0,0 +1,37 @@
package com.fastbee.base.core;
import com.fastbee.common.core.protocol.Message;
import com.fastbee.base.session.Session;
/**
* 消息拦截器
* @author bill
*/
public interface HandlerInterceptor<T extends Message> {
/**
* 未匹配到对应的Handle消息处理
*/
T notSupported(T request, Session session);
/**
* 调用之前
* 处理消息类型匹配
*/
boolean beforeHandle(T request, Session session);
/**
* 需要应答设备,在这里执行
* 调用之后返回值为void的 */
T successful(T request, Session session);
/** 调用之后,有返回值的 */
void afterHandle(T request, T response, Session session);
/**
* 报错应答方法
* 调用之后抛出异常的
*/
T exceptional(T request, Session session, Exception e);
}

View File

@@ -0,0 +1,12 @@
package com.fastbee.base.core;
import com.fastbee.base.core.hanler.BaseHandler;
/**
* 消息处理接口
* @author bill
*/
public interface HandlerMapping {
BaseHandler getHandler(int messageId);
}

View File

@@ -0,0 +1,23 @@
package com.fastbee.base.core;
import com.fastbee.base.core.annotation.Node;
import org.springframework.beans.BeansException;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import java.util.Map;
/**
*
* @author bill
*/
public class SpringHandlerMapping extends AbstractHandlerMapping implements ApplicationContextAware {
@Override
public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
Map<String, Object> endpoints = applicationContext.getBeansWithAnnotation(Node.class);
for (Object bean : endpoints.values()) {
super.registerHandlers(bean);
}
}
}

View File

@@ -0,0 +1,15 @@
package com.fastbee.base.core.annotation;
import java.lang.annotation.*;
/**
* 异步处理设备数据
*
* @author bill
*/
@Target({ElementType.METHOD})
@Retention(RetentionPolicy.RUNTIME)
@Documented
public @interface Async {
}

View File

@@ -0,0 +1,27 @@
package com.fastbee.base.core.annotation;
import java.lang.annotation.*;
/**
* 多线程异步处理设备数据,新建线程组处理
*
* @author bill
*/
@Target({ElementType.METHOD})
@Retention(RetentionPolicy.RUNTIME)
@Documented
public @interface AsyncBatch {
/*批量处理的最大消息数*/
int maxMessageSize() default 5000;
/*线程数*/
int poolSize() default 2;
/*最大等待时间*/
int maxWaitTime() default 1000;
/*最小处理消息数*/
int minMessageSize() default 100;
}

View File

@@ -0,0 +1,14 @@
package com.fastbee.base.core.annotation;
import java.lang.annotation.*;
/**
* 消息节点
* @author bill
*/
@Target({ElementType.TYPE})
@Retention(RetentionPolicy.RUNTIME)
@Documented
public @interface Node {
}

View File

@@ -0,0 +1,17 @@
package com.fastbee.base.core.annotation;
import java.lang.annotation.*;
/**
* 字段映射
* @author bill
*/
@Target({ElementType.METHOD})
@Retention(RetentionPolicy.RUNTIME)
@Documented
public @interface PakMapping {
int[] types();
String desc() default "";
}

View File

@@ -0,0 +1,128 @@
package com.fastbee.base.core.hanler;
import com.fastbee.common.core.protocol.Message;
import com.fastbee.base.session.Session;
import com.fastbee.base.util.VirtualList;
import com.fastbee.common.exception.ServiceException;
import io.netty.util.concurrent.DefaultThreadFactory;
import lombok.extern.slf4j.Slf4j;
import java.lang.reflect.Method;
import java.util.List;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
/**
* 异步批量处理报文
* @author bill
*/
@Slf4j
public class AsyncBatchHandler extends BaseHandler{
/*消息处理队列*/
private final ConcurrentLinkedQueue<Message> queue;
/*线程池*/
private final ThreadPoolExecutor executor;
private final int poolSize;
private final int maxEventSize;
private final int maxWait;
private final int warningLines;
public AsyncBatchHandler(Object target, Method targetMethod, String desc, int poolSize, int maxEventSize, int maxWait) {
super(target, targetMethod, desc);
Class<?>[] parameterTypes = targetMethod.getParameterTypes();
if (parameterTypes.length >1){
throw new ServiceException("参数列表过长");
}
if (!parameterTypes[0].isAssignableFrom(List.class)){
throw new ServiceException("参数不是List类型");
}
this.poolSize = poolSize;
this.maxEventSize = maxEventSize;
this.maxWait = maxWait;
this.warningLines = maxEventSize * poolSize * 50;
this.queue = new ConcurrentLinkedQueue<>();
this.executor = new ThreadPoolExecutor(this.poolSize,this.poolSize,1000L, TimeUnit.MILLISECONDS
,new LinkedBlockingQueue<>(500),new DefaultThreadFactory(targetMethod.getName()));
for (int i = 0; i < poolSize; i++) {
boolean start = i == 0;
executor.execute(()->{
try {
startInternal(start);
}catch (Exception e){
log.error("线程池处理数据出错",e);
}
});
}
}
@Override
public <T extends Message> T invoke(T request, Session session) throws Exception {
queue.offer(request);
return null;
}
public void startInternal(boolean master) {
Message[] array = new Message[maxEventSize];
long logtime = 0;
long starttime = 0;
for (; ; ) {
Message temp;
int i = 0;
while ((temp = queue.poll()) != null) {
array[i++] = temp;
if (i >= maxEventSize) {
break;
}
}
if (i > 0) {
starttime = System.currentTimeMillis();
try {
targetMethod.invoke(targetObject, new VirtualList<>(array, i));
} catch (Exception e) {
log.warn(targetMethod.getName(), e);
}
long time = System.currentTimeMillis() - starttime;
if (time > 1000L) {
log.warn("线程池处理数据耗时:{}ms,共{}条记录", time, i);
}
}
if (i < maxEventSize) {
try {
for (int j = 0; j < i; j++) {
array[j] = null;
}
Thread.sleep(maxWait);
} catch (InterruptedException e) {
log.error("sleep error!");
}
} else if (master) {
if (logtime < starttime) {
logtime = starttime + 5000L;
int size = queue.size();
if (size > warningLines) {
log.warn("线程池队列已满, size:{}", size);
}
}
}
}
}
}

View File

@@ -0,0 +1,86 @@
package com.fastbee.base.core.hanler;
import com.fastbee.common.core.protocol.Message;
import com.fastbee.base.session.Session;
import sun.reflect.generics.reflectiveObjects.ParameterizedTypeImpl;
import java.lang.reflect.Method;
import java.lang.reflect.Type;
/**
* 基础处理类
*
* @author bill
*/
public abstract class BaseHandler {
public static final int MESSAGE = 0;
public static final int SESSION = 1;
public final Object targetObject;
public final Method targetMethod;
public final int[] parameterTypes;
public final boolean returnVoid;
public final boolean async;
public final String desc;
public BaseHandler(Object target, Method targetMethod, String desc) {
this(target, targetMethod, desc, false);
}
public BaseHandler(Object targetObject, Method targetMethod, String desc, boolean async) {
this.targetObject = targetObject;
this.targetMethod = targetMethod;
this.returnVoid = targetMethod.getReturnType().isAssignableFrom(Void.TYPE);
this.async = async;
if (desc == null || desc.isEmpty())
desc = targetMethod.getName();
this.desc = desc;
Type[] types = targetMethod.getGenericParameterTypes();
int[] parameterTypes = new int[types.length];
try {
for (int i = 0; i < types.length; i++) {
Type type = types[i];
Class<?> clazz;
if (type instanceof ParameterizedTypeImpl) {
clazz = (Class<?>) ((ParameterizedTypeImpl) type).getActualTypeArguments()[0];
} else {
clazz = (Class<?>) type;
}
if (Message.class.isAssignableFrom(clazz)) {
parameterTypes[i] = MESSAGE;
} else if (Session.class.isAssignableFrom(clazz)) {
parameterTypes[i] = SESSION;
}
}
} catch (Exception e) {
throw new RuntimeException(e);
}
this.parameterTypes = parameterTypes;
}
public <T extends Message> T invoke(T request, Session session) throws Exception {
Object[] args = new Object[parameterTypes.length];
for (int i = 0; i < parameterTypes.length; i++) {
int type = parameterTypes[i];
switch (type) {
case BaseHandler.MESSAGE:
args[i] = request;
break;
case BaseHandler.SESSION:
args[i] = session;
break;
}
}
return (T) targetMethod.invoke(targetObject, args);
}
@Override
public String toString() {
return desc;
}
}

View File

@@ -0,0 +1,23 @@
package com.fastbee.base.core.hanler;
import com.fastbee.common.core.protocol.Message;
import com.fastbee.base.session.Session;
import java.lang.reflect.Method;
/**
* 同步处理报文
* @author bill
*/
public class SyncHandler extends BaseHandler{
public SyncHandler(Object target, Method targetMethod, String desc,boolean async) {
super(target, targetMethod, desc, async);
}
@Override
public <T extends Message> T invoke(T request, Session session) throws Exception {
return super.invoke(request, session);
}
}

View File

@@ -0,0 +1,12 @@
package com.fastbee.base.core.model;
/**
* 消息流水号响应
* @author gsb
* @date 2022/11/7 10:19
*/
public interface Response {
/**应答消息流水号*/
int getResponseSerialNo();
}

View File

@@ -0,0 +1,19 @@
package com.fastbee.base.model;
import lombok.Data;
/**
* @author gsb
* @date 2023/3/9 10:07
*/
@Data
public class DeviceMsg {
protected String clientId;
protected Long deviceId;
private int protocolVersion;
private Long productId;
}

View File

@@ -0,0 +1,16 @@
package com.fastbee.base.model;
import com.fastbee.base.session.Session;
/**
* @author gsb
* @date 2023/3/9 10:03
*/
public enum SessionKey {
DeviceMsg;
public static DeviceMsg getDeviceMsg(Session session){
return (DeviceMsg)session.getAttribute(SessionKey.DeviceMsg);
}
}

View File

@@ -0,0 +1,60 @@
package com.fastbee.base.service;
import com.fastbee.base.session.Session;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
/**
* 服务会话存储接口
* @author gsb
* @date 2022/10/14 14:16
*/
public interface ISessionStore {
/**
* session 会话存储
*
* @param clientId: 客户端标识
* @param session: session会话
*/
void storeSession(String clientId, Session session);
/**
* 根据客户端标识获取相应会话
*
* @param clientId: 客户端标识
*/
Session getSession(String clientId);
/**
* 清除历史会话状态
*
* @param clientId: 客户端标识
*/
void cleanSession(String clientId);
/**
* 根据客户端标识查看是否存在该会话
*
* @param clientId:
*/
boolean containsKey(String clientId);
/**
* 获取集合
* @return MAP
*/
ConcurrentHashMap<String, Session> getSessionMap();
/**
* map分页从1开始
*
* @param sourceMap 分页数据
* @param pageSize 页面大小
* @param currentPage 当前页面
*/
public Map<String, Session> listPage(Map<String, Session> sourceMap, int pageSize, int currentPage);
}

View File

@@ -0,0 +1,104 @@
package com.fastbee.base.service.impl;
import com.fastbee.base.service.ISessionStore;
import com.fastbee.base.session.Session;
import org.springframework.stereotype.Service;
import java.util.LinkedHashMap;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicInteger;
/**
* 内存存储服务会话
*
* @author gsb
* @date 2022/10/14 14:18
*/
@Service
public class SessionStoreImpl implements ISessionStore {
/*session存储集合*/
private final ConcurrentHashMap<String, Session> sessionMap = new ConcurrentHashMap<>();
/**
* MQTT会话存储
*
* @param clientId: 客户端标识
* @param session: MQTT会话
*/
@Override
public void storeSession(String clientId, Session session) {
sessionMap.put(clientId, session);
}
/**
* 根据客户端标识获取相应会话
*
* @param clientId: 客户端标识
*/
@Override
public Session getSession(String clientId) {
return sessionMap.get(clientId);
}
/**
* 清除历史会话状态
*
* @param clientId: 客户端标识
*/
@Override
public void cleanSession(String clientId) {
sessionMap.remove(clientId);
}
/**
* 根据客户端标识查看是否存在该会话
*
* @param clientId:
*/
@Override
public boolean containsKey(String clientId) {
return sessionMap.containsKey(clientId);
}
/**
* 获取集合
* @return MAP
*/
@Override
public ConcurrentHashMap<String, Session> getSessionMap(){
return sessionMap;
}
/**
* map分页从1开始
*
* @param sourceMap 分页数据
* @param pageSize 页面大小
* @param currentPage 当前页面
*/
@Override
public Map<String, Session> listPage(Map<String, Session> sourceMap, int pageSize, int currentPage) {
Map<String, Session> map = new LinkedHashMap<>();
if (sourceMap.size() > 0) {
AtomicInteger flag = new AtomicInteger(0);
AtomicInteger size = new AtomicInteger(0);
int currIdx = (currentPage > 1 ? (currentPage - 1) * pageSize : 0);
sourceMap.forEach((ass, list_km) -> {
if (flag.get() >= currIdx) {
if (size.get() < pageSize) {
map.put(ass, list_km);
} else {
return;
}
size.getAndIncrement();
}
flag.getAndIncrement();
});
}
return map;
}
}

View File

@@ -0,0 +1,74 @@
package com.fastbee.base.session;
import com.fastbee.common.core.protocol.Message;
import io.netty.buffer.ByteBuf;
import io.netty.channel.socket.DatagramPacket;
/**
* 报文消息的包装
* @author bill
*/
public abstract class Packet {
/*session*/
public final Session session;
/*基础消息*/
public Message message;
/*报文缓存buf*/
public ByteBuf byteBuf;
public static Packet of(Session session, Message message) {
if (session.isUdp()) {
return new UDP(session, message, null);
}
return new TCP(session, message, message.getPayload());
}
public static Packet of(Session session, ByteBuf message) {
if (session.isUdp()) {
return new UDP(session, null, message);
}
return new TCP(session, null, message);
}
private Packet(Session session, Message message, ByteBuf byteBuf) {
this.session = session;
this.message = message;
this.byteBuf = byteBuf;
}
public Packet replace(Message message) {
this.message = message;
return this;
}
public ByteBuf take() {
ByteBuf temp = this.byteBuf;
this.byteBuf = null;
return temp;
}
public abstract Object wrap(ByteBuf byteBuf);
private static class TCP extends Packet {
private TCP(Session session, Message message, ByteBuf byteBuf) {
super(session, message, byteBuf);
}
@Override
public Object wrap(ByteBuf byteBuf) {
return byteBuf;
}
}
private static class UDP extends Packet {
private UDP(Session session, Message message, ByteBuf byteBuf) {
super(session, message, byteBuf);
}
@Override
public Object wrap(ByteBuf byteBuf) {
return new DatagramPacket(byteBuf, session.remoteAddress());
}
}
}

View File

@@ -0,0 +1,286 @@
package com.fastbee.base.session;
import com.fastbee.common.core.protocol.Message;
import com.fastbee.common.enums.ServerType;
import com.fastbee.common.exception.ServiceException;
import com.fastbee.common.utils.DateUtils;
import com.fastbee.common.utils.gateway.mq.Topics;
import com.fastbee.base.core.model.Response;
import com.fasterxml.jackson.annotation.JsonFormat;
import io.netty.buffer.ByteBuf;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.mqtt.MqttMessageType;
import io.netty.handler.codec.mqtt.MqttVersion;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import reactor.core.publisher.Mono;
import reactor.core.publisher.MonoSink;
import java.net.InetSocketAddress;
import java.util.*;
import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.Function;
/**
* 会话
*
* @Author guanshubiao
* @Date 2022/9/12 20:22
*/
@Data
@Slf4j
public class Session {
private boolean udp;
private Function<Session, Boolean> remover;
protected Channel channel;
/**
* 客户端id
*/
private String clientId;
private String productId;
private SessionManager sessionManager;
private InetSocketAddress remoteAddress;
private String remoteAddressStr;
private long creationTime;
private long lastAccessTime;
private Map<Object, Object> attributes;
private String sessionId;
//原子计数器,报文没有消息流水号的,充当流水号
private AtomicInteger serialNo = new AtomicInteger(0);
private int keepAlive = 60;
/*mqtt版本号*/
private MqttVersion version;
/*是否清楚会话*/
private Boolean cleanSession = false;
/*mqtt账号*/
private String username;
/*是否链接*/
private Boolean connected = false;
/*mqtt消息类型*/
private MqttMessageType mqttMessageType;
private int keepAliveMax = 120;
/*主题*/
private String topicName;
/*Channel处理类上下文*/
private ChannelHandlerContext handlerContext;
private List<Topics> topics;
private Integer topicCount;
@JsonFormat(pattern = "yyyy-MM-dd HH:mm:ss")
private Date connected_at;
private String ip;
/*服务协议类型 MQTTTCP UDP COAP*/
private ServerType serverType;
public Session(){
}
public Session(SessionManager sessionManager,
Channel channel,
InetSocketAddress remoteAddress,
Function<Session, Boolean> remover,
boolean udp,
ServerType serverType) {
this.channel = channel;
this.creationTime = DateUtils.getTimestamp();
this.lastAccessTime = creationTime;
this.sessionManager = sessionManager;
this.remoteAddress = remoteAddress;
this.remoteAddressStr = remoteAddress.toString();
this.remover = remover;
this.udp = udp;
this.serverType = serverType;
this.ip = remoteAddress.getHostName();
this.connected = true;
this.connected_at = DateUtils.getNowDate();
if (sessionManager != null && sessionManager.getSessionKeys() != null) {
this.attributes = new EnumMap(sessionManager.getSessionKeys());
} else {
this.attributes = new TreeMap<>();
}
}
/**
* 判断设备是否已经注册上线
*/
public boolean isRegistered() {
return clientId != null;
}
/*设备端注册*/
public void register(Message message) {
register(message.getClientId(), message);
}
public void register(String clientId, Message message) {
//已经注册,不再发送注册包数据
if (isRegistered()){
return;
}
if (clientId == null) {
throw new ServiceException("客户端注册clientId不能为空");
}
this.clientId = clientId.toUpperCase();
if (sessionManager != null) {
sessionManager.add(this);
}
}
public Object getAttribute(Object name) {
return attributes.get(name);
}
public void setAttribute(Object name, Object value) {
attributes.put(name, value);
}
/**
* 获取最后上线时间
*/
public long access() {
return lastAccessTime = DateUtils.getTimestamp();
}
/**
* 获取远程端口
*/
public InetSocketAddress remoteAddress() {
return remoteAddress;
}
/**
* 获取流水号
*/
public int nextSerialNO() {
int current;
int next;
do {
current = serialNo.get();
next = current > 0xffff ? 0 : current;
} while (!serialNo.compareAndSet(current, next + 1));
return next;
}
/**
* 处理离线方法
*/
public void invalidate() {
if (isRegistered() && sessionManager != null) {
sessionManager.remove(this);
}
remover.apply(this);
}
public boolean isUdp() {
return udp;
}
private final Map<String, MonoSink> topicSubscribers = new HashMap<>();
private static final Mono rejected = Mono.error(new RejectedExecutionException("设备端暂无响应"));
/**
* 异步发送通知类消息
* 同步发送 mono.block()
* 订阅回调 mono.doOnSuccess(处理成功).doOnError(处理异常).subscribe(开始订阅)
*/
public Mono<Void> notify(Message message) {
return mono(channel.writeAndFlush(Packet.of(this, message)));
}
public Mono<Void> notify(ByteBuf message) {
return mono(channel.writeAndFlush(Packet.of(this, message)));
}
public static Mono<Void> mono(ChannelFuture channelFuture) {
return Mono.create(sink -> channelFuture.addListener(future -> {
if (future.isSuccess()) {
sink.success();
} else {
sink.error(future.cause());
}
}));
}
/**
* 异步发送消息,接收响应
* 同步接收 mono.block()
* 订阅回调 mono.doOnSuccess(处理成功).doOnError(处理异常).subscribe(开始订阅)
*/
public <T> Mono<T> request(Message message, Class<T> responseClass) {
String key = requestKey(message, responseClass);
Mono<T> subscribe = this.subscribe(key);
if (subscribe == null) {
return rejected;
}
ChannelFuture channelFuture = channel.writeAndFlush(Packet.of(this, message));
return Mono.create(sink -> channelFuture.addListener(future -> {
if (future.isSuccess()) {
sink.success(future);
} else {
sink.error(future.cause());
}
})).then(subscribe).doFinally(signal -> unsubscribe(key));
}
/**
* 消息响应
*/
public boolean response(Message message){
MonoSink monoSink = topicSubscribers.get(responseKey(message));
if (monoSink != null){
monoSink.success(message);
return true;
}
return false;
}
/**
* 订阅
*/
private Mono subscribe(String key) {
synchronized (topicSubscribers) {
if (!topicSubscribers.containsKey(key)) {
return Mono.create(sink -> topicSubscribers.put(key, sink));
}
}
return null;
}
/**
* 取消订阅
*/
private void unsubscribe(String key) {
topicSubscribers.remove(key);
}
/*生成流水号*/
private static String requestKey(Message request, Class responseClass) {
String className = responseClass.getName();
if (Response.class.isAssignableFrom(responseClass)) {
String serNo = request.getSerNo();
return new StringBuffer(34).append(className).append('.').append(serNo).toString();
}
return className;
}
/*返回流水号*/
private static String responseKey(Object response) {
String className = response.getClass().getName();
if (response instanceof Response) {
int serialNo = ((Response) response).getResponseSerialNo();
return new StringBuffer(34).append(className).append('.').append(serialNo).toString();
}
return className;
}
}

View File

@@ -0,0 +1,26 @@
package com.fastbee.base.session;
/**
* session监听
* @author gsb
* @date 2022/11/7 8:57
*/
public interface SessionListener {
/** 客户端建立连接 */
default void sessionCreated(Session session) {
}
/** 客户端完成注册或鉴权 */
default void sessionRegistered(Session session) {
}
/**
* 客户端注销或离线
*/
default void sessionDestroyed(Session session) {
}
}

View File

@@ -0,0 +1,125 @@
package com.fastbee.base.session;
import com.fastbee.common.enums.ServerType;
import com.fastbee.common.utils.spring.SpringUtils;
import com.fastbee.base.service.ISessionStore;
import io.netty.channel.Channel;
import lombok.extern.slf4j.Slf4j;
import java.net.InetSocketAddress;
import java.util.Collection;
import java.util.Locale;
import java.util.concurrent.ConcurrentHashMap;
import java.util.function.Function;
/**
* session管理
* @author gsb
* @date 2022/11/7 8:55
*/
@Slf4j
public class SessionManager {
private final Class<? extends Enum> sessionKeys;
private final SessionListener sessionListener;
/*Session会话存储*/
private static ISessionStore sessionStore = SpringUtils.getBean(ISessionStore.class);
public SessionManager(){
this(null,null);
}
public SessionManager(SessionListener sessionListener){
this(null,sessionListener);
}
public SessionManager(Class<? extends Enum> sessionKeys, SessionListener sessionListener){
this.sessionKeys = sessionKeys;
this.sessionListener = sessionListener;
}
/**
* 获取Session
*/
public Session getSession(String clientId) {
return sessionStore.getSession(clientId);
}
/**
* 获取所有session
*/
public Collection<Session> all(){
return sessionStore.getSessionMap().values();
}
/**
* 新建session TCP
* @return
*/
public Session newInstance(Channel channel){
InetSocketAddress sender = (InetSocketAddress) channel.remoteAddress();
Session session = new Session(this, channel, sender, s -> {
channel.close();
return true;
}, false, ServerType.TCP);
if (sessionListener != null) {
try {
sessionListener.sessionCreated(session);
} catch (Exception e) {
log.error("sessionCreated", e);
}
}
return session;
}
/**
* 新建session UDP
*/
public Session newInstance(Channel channel, InetSocketAddress sender, Function<Session, Boolean> remover) {
Session session = new Session(this, channel, sender, remover, true,ServerType.UDP);
if (sessionListener != null) {
try {
sessionListener.sessionCreated(session);
} catch (Exception e) {
log.error("sessionCreated", e);
}
}
return session;
}
/**
* 设备端离线
*/
protected void remove(Session session){
sessionStore.cleanSession(session.getClientId());
if (null != sessionListener){
try {
//设备状态业务处理
sessionListener.sessionDestroyed(session);
}catch (Exception e){
log.error("设备端离线异常",e);
}
}
}
/**
* 设备端上线
*/
protected void add(Session session){
sessionStore.storeSession(session.getClientId().toUpperCase(),session);
if (null != sessionListener){
try {
sessionListener.sessionRegistered(session);
}catch (Exception e){
log.error("设备端注册",e);
}
}
}
public Class<? extends Enum> getSessionKeys(){
return sessionKeys;
}
}

View File

@@ -0,0 +1,42 @@
package com.fastbee.base.util;
import com.fastbee.base.session.Session;
import io.netty.channel.Channel;
import io.netty.util.AttributeKey;
/**
* channel和pipeline桥梁Attribute信息构建
* @author gsb
* @date 2022/10/7 18:40
*/
public class AttributeUtils {
/*存储客户端连接信息*/
private static final AttributeKey<Object> SESSION_KEY = AttributeKey.valueOf("session");
/*存储客户端id*/
private static final AttributeKey<Object> CLIENT_ID_KEY = AttributeKey.valueOf("clientId");
/*添加session*/
public static void setSession(Channel channel, Session session){
channel.attr(SESSION_KEY).set(session);
}
/*获取session*/
public static Session getSession(Channel channel){
return (Session) channel.attr(SESSION_KEY).get();
}
/*添加clientId*/
public static void setClientId(Channel channel, String clientId){
channel.attr(CLIENT_ID_KEY).set(clientId);
}
/*获取clientId*/
public static String getClientId(Channel channel){
return (String) channel.attr(CLIENT_ID_KEY).get();
}
}

View File

@@ -0,0 +1,106 @@
package com.fastbee.base.util;
import java.io.File;
import java.lang.annotation.Annotation;
import java.net.JarURLConnection;
import java.net.URL;
import java.util.Enumeration;
import java.util.LinkedList;
import java.util.List;
import java.util.jar.JarEntry;
import java.util.jar.JarFile;
/**
* 类操作工具
* @author bill
*/
public class ClassUtils {
public static List<Class> getClassList(String packageName, Class<? extends Annotation> annotationClass) {
List<Class> classList = getClassList(packageName);
classList.removeIf(next -> !next.isAnnotationPresent(annotationClass));
return classList;
}
public static List<Class> getClassList(String packageName) {
List<Class> classList = new LinkedList<>();
String path = packageName.replace(".", "/");
try {
Enumeration<URL> urls = ClassUtils.getClassLoader().getResources(path);
while (urls.hasMoreElements()) {
URL url = urls.nextElement();
if (url != null) {
String protocol = url.getProtocol();
if (protocol.equals("file")) {
addClass(classList, url.toURI().getPath(), packageName);
} else if (protocol.equals("jar")) {
JarURLConnection jarURLConnection = (JarURLConnection) url.openConnection();
JarFile jarFile = jarURLConnection.getJarFile();
Enumeration<JarEntry> jarEntries = jarFile.entries();
while (jarEntries.hasMoreElements()) {
JarEntry jarEntry = jarEntries.nextElement();
String entryName = jarEntry.getName();
if (entryName.startsWith(path) && entryName.endsWith(".class")) {
String className = entryName.substring(0, entryName.lastIndexOf(".")).replaceAll("/", ".");
addClass(classList, className);
}
}
}
}
}
} catch (Exception e) {
throw new RuntimeException("Initial class error!");
}
return classList;
}
private static void addClass(List<Class> classList, String packagePath, String packageName) {
try {
File[] files = new File(packagePath).listFiles(file -> (file.isDirectory() || file.getName().endsWith(".class")));
if (files != null)
for (File file : files) {
String fileName = file.getName();
if (file.isFile()) {
String className = fileName.substring(0, fileName.lastIndexOf("."));
if (packageName != null) {
className = packageName + "." + className;
}
addClass(classList, className);
} else {
String subPackagePath = fileName;
if (packageName != null) {
subPackagePath = packagePath + "/" + subPackagePath;
}
String subPackageName = fileName;
if (packageName != null) {
subPackageName = packageName + "." + subPackageName;
}
addClass(classList, subPackagePath, subPackageName);
}
}
} catch (Exception e) {
throw new RuntimeException(e);
}
}
private static void addClass(List<Class> classList, String className) {
classList.add(loadClass(className, false));
}
public static Class loadClass(String className, boolean isInitialized) {
try {
return Class.forName(className, isInitialized, getClassLoader());
} catch (ClassNotFoundException e) {
throw new RuntimeException(e);
}
}
public static ClassLoader getClassLoader() {
return Thread.currentThread().getContextClassLoader();
}
}

View File

@@ -0,0 +1,51 @@
package com.fastbee.base.util;
import lombok.Data;
import java.util.concurrent.ConcurrentHashMap;
/**
* concurrentMap存储类
* @author bill
*/
@Data
public abstract class ConcurrentStorage<K,V> implements Storage<K,V>{
private volatile ConcurrentHashMap<K,V> map;
public ConcurrentStorage(){
this(new ConcurrentHashMap<K,V>());
}
public ConcurrentStorage(ConcurrentHashMap<K,V> map){
this.map = map;
}
@Override
public V push(K key, V value) {
return map.put(key,value);
}
@Override
public V pop(K key) {
return map.get(key);
}
@Override
public V remove(K key) {
return map.remove(key);
}
@Override
public boolean isContains(Object key) {
return map.containsKey(key);
}
@Override
public Object getStorage() {
return this.map;
}
}

View File

@@ -0,0 +1,30 @@
package com.fastbee.base.util;
import com.fastbee.common.core.mq.DeviceStatusBo;
import com.fastbee.common.enums.DeviceStatus;
import com.fastbee.common.utils.DateUtils;
import io.netty.channel.Channel;
import java.net.InetSocketAddress;
/**
* 设备信息工具类
* @author bill
*/
public class DeviceUtils {
/*构造返回MQ的设备状态model*/
public static DeviceStatusBo buildStatusMsg(Channel channel, String clientId, DeviceStatus status, String ip){
InetSocketAddress address = (InetSocketAddress) channel.remoteAddress();
return DeviceStatusBo.builder()
.serialNumber(clientId)
.status(status)
.ip(ip)
.hostName(address.getHostName())
.port(address.getPort())
.timestamp(DateUtils.getNowDate()).build();
}
}

View File

@@ -0,0 +1,48 @@
package com.fastbee.base.util;
import java.util.concurrent.atomic.AtomicInteger;
/**
* @author bill
*/
public class Stopwatch {
private final AtomicInteger count = new AtomicInteger();
private final Thread thread;
public Stopwatch start() {
this.thread.start();
return this;
}
public int increment() {
return count.incrementAndGet();
}
public Stopwatch() {
thread = new Thread(() -> {
long start;
while (true) {
if (count.get() > 0) {
start = System.currentTimeMillis();
break;
}
try {
Thread.sleep(1L);
} catch (Exception e) {
}
}
while (true) {
try {
Thread.sleep(2000L);
} catch (Exception e) {
}
int num = count.get();
long time = (System.currentTimeMillis() - start) / 1000;
System.out.println(time + "\t" + num + "\t" + num / time);
}
});
thread.setName(Thread.currentThread().getName() + "-c");
thread.setPriority(Thread.MIN_PRIORITY);
thread.setDaemon(true);
}
}

View File

@@ -0,0 +1,18 @@
package com.fastbee.base.util;
/**
* 存储管理
* @author bill
*/
public interface Storage<K,V> {
V push(K key, V value);
V pop(K key);
V remove(K key);
boolean isContains(K key);
Object getStorage();
}

View File

@@ -0,0 +1,101 @@
package com.fastbee.base.util;
import java.io.Serializable;
import java.util.*;
import java.util.function.Consumer;
import java.util.function.UnaryOperator;
/**
* @author bill
*/
public class VirtualList<E> extends AbstractList<E> implements RandomAccess, Serializable {
private final E[] elementData;
private final int size;
public VirtualList(E[] array, int length) {
this.elementData = array;
this.size = length;
}
@Override
public int size() {
return size;
}
@Override
public Object[] toArray() {
return elementData.clone();
}
@Override
@SuppressWarnings("unchecked")
public <T> T[] toArray(T[] a) {
if (a.length < size)
return Arrays.copyOf(this.elementData, size,
(Class<? extends T[]>) a.getClass());
System.arraycopy(this.elementData, 0, a, 0, size);
if (a.length > size)
a[size] = null;
return a;
}
@Override
public E get(int index) {
return elementData[index];
}
@Override
public E set(int index, E element) {
E oldValue = elementData[index];
elementData[index] = element;
return oldValue;
}
@Override
public int indexOf(Object o) {
E[] a = this.elementData;
if (o == null) {
for (int i = 0; i < size; i++)
if (a[i] == null)
return i;
} else {
for (int i = 0; i < size; i++)
if (o.equals(a[i]))
return i;
}
return -1;
}
@Override
public boolean contains(Object o) {
return indexOf(o) != -1;
}
@Override
public Spliterator<E> spliterator() {
return Spliterators.spliterator(elementData, 0, size, Spliterator.ORDERED);
}
@Override
public void forEach(Consumer<? super E> action) {
Objects.requireNonNull(action);
for (int i = 0; i < size; i++) {
action.accept(elementData[i]);
}
}
@Override
public void replaceAll(UnaryOperator<E> operator) {
Objects.requireNonNull(operator);
E[] a = this.elementData;
for (int i = 0; i < size; i++) {
a[i] = operator.apply(a[i]);
}
}
@Override
public void sort(Comparator<? super E> c) {
Arrays.sort(elementData, 0, size, c);
}
}

View File

@@ -0,0 +1,35 @@
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<parent>
<artifactId>fastbee-server</artifactId>
<groupId>com.fastbee</groupId>
<version>3.8.5</version>
</parent>
<artifactId>boot-strap</artifactId>
<description>网关服务启动模块</description>
<dependencies>
<dependency>
<groupId>com.fastbee</groupId>
<artifactId>mqtt-broker</artifactId>
</dependency>
<dependency>
<groupId>com.fastbee</groupId>
<artifactId>fastbee-protocol-collect</artifactId>
</dependency>
<dependency>
<groupId>com.fastbee</groupId>
<artifactId>base-server</artifactId>
<version>3.8.5</version>
<scope>compile</scope>
</dependency>
</dependencies>
</project>

View File

@@ -0,0 +1,70 @@
package com.fastbee.bootstrap.mqtt;
import com.fastbee.mqtt.server.MqttServer;
import com.fastbee.mqtt.server.WebSocketServer;
import com.fastbee.server.Server;
import com.fastbee.server.config.NettyConfig;
import com.fastbee.common.enums.ServerType;
import lombok.Data;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.core.annotation.Order;
/**
* MQTT-BROKER启动
* @author gsb
* @date 2022/9/17 17:25
*/
@Order(10)
@Configuration
@ConfigurationProperties(value = "server.broker")
@Data
public class MQTTBootStrap {
@Autowired
private MqttServer mqttServer;
@Autowired
private WebSocketServer webSocketServer;
/*服务器集群节点*/
private String brokerNode;
/*端口*/
private int port;
/*心跳时间*/
private int keepAlive;
/*webSocket端口*/
private int websocketPort;
/*webSocket路由*/
private String websocketPath;
/**
* 启动mqttBroker
* @return server
*/
@ConditionalOnProperty(value = "server.broker.enabled", havingValue = "true")
@Bean(initMethod = "start", destroyMethod = "stop")
public Server mqttBroker() {
return NettyConfig.custom()
.setIdleStateTime(0,0,keepAlive)
.setName(ServerType.MQTT.getDes())
.setType(ServerType.MQTT)
.setPort(port)
.setServer(mqttServer)
.build();
}
@ConditionalOnProperty(value = "server.broker.enabled", havingValue = "true")
@Bean(initMethod = "start",destroyMethod = "stop")
public Server webSocket(){
return NettyConfig.custom()
.setIdleStateTime(0,0,keepAlive)
.setName(ServerType.WEBSOCKET.getDes())
.setType(ServerType.WEBSOCKET)
.setPort(websocketPort)
.setServer(webSocketServer)
.build();
}
}

View File

@@ -0,0 +1,25 @@
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<parent>
<artifactId>fastbee-server</artifactId>
<groupId>com.fastbee</groupId>
<version>3.8.5</version>
</parent>
<artifactId>iot-server-core</artifactId>
<dependencies>
<dependency>
<groupId>com.fastbee</groupId>
<artifactId>base-server</artifactId>
<scope>compile</scope>
</dependency>
<dependency>
<groupId>com.fastbee</groupId>
<artifactId>fastbee-mq</artifactId>
</dependency>
</dependencies>
</project>

View File

@@ -0,0 +1,77 @@
package com.fastbee.server;
import com.fastbee.server.config.NettyConfig;
import io.netty.bootstrap.AbstractBootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.EventLoopGroup;
import lombok.NoArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import java.util.concurrent.ExecutorService;
/**
* 基础服务器启动类
*
* @Author guanshubiao
* @Date 2022/9/12 20:22
*/
@Slf4j
@NoArgsConstructor
public abstract class Server {
protected EventLoopGroup bossGroup;
protected EventLoopGroup workerGroup;
protected ExecutorService businessService;
protected boolean isRunning;
public NettyConfig config;
protected Server(NettyConfig config){
this.config = config;
}
/*初始化方法*/
protected abstract AbstractBootstrap initialize();
public synchronized boolean start() {
if (isRunning) {
log.warn("=>服务:{},在端口:{},已经运行", config.name, config.port);
return isRunning;
}
AbstractBootstrap bootstrap = initialize();
ChannelFuture future = bootstrap.bind(config.port).awaitUninterruptibly();
future.channel().closeFuture().addListener(event -> {
if (isRunning) {
stop();
}
});
if (isRunning = future.isSuccess()) {
log.info("=>服务:{},在端口:{},启动成功!", config.name, config.port);
return isRunning;
}
if (future.cause() != null) {
log.error("服务启动失败", future.cause());
}
return isRunning;
}
public synchronized void stop() {
isRunning = false;
bossGroup.shutdownGracefully();
if (workerGroup != null) {
workerGroup.shutdownGracefully();
}
if (businessService != null) {
businessService.shutdown();
}
log.warn("=>服务:{},在端口:{},已经停止!", config.name, config.port);
}
}

View File

@@ -0,0 +1,254 @@
package com.fastbee.server.config;
import com.fastbee.common.constant.FastBeeConstant;
import com.fastbee.common.enums.ServerType;
import com.fastbee.server.Server;
import com.fastbee.base.codec.Delimiter;
import com.fastbee.base.codec.LengthField;
import com.fastbee.base.codec.MessageDecoder;
import com.fastbee.base.codec.MessageEncoder;
import com.fastbee.base.core.HandlerInterceptor;
import com.fastbee.base.core.HandlerMapping;
import com.fastbee.base.session.SessionManager;
import io.netty.util.NettyRuntime;
import io.netty.util.internal.ObjectUtil;
/**
* 基础配置类
* @Author guanshubiao
* @Date 2022/9/12 20:22
*/
public class NettyConfig {
public final int workerCore;
/*boss线程核数*/
public final int businessCore;
/*读空闲时间*/
public final int readerIdleTime;
/*写空闲时间*/
public final int writerIdleTime;
/*读写空闲时间*/
public final int allIdleTime;
/*端口*/
public final Integer port;
/*TCP/UDP数据最大长度限定*/
public final Integer maxFrameLength;
/*基础编码*/
public final MessageDecoder decoder;
/*基础解码*/
public final MessageEncoder encoder;
public final Delimiter[] delimiters;
public final LengthField lengthField;
public final HandlerMapping handlerMapping;
public final HandlerInterceptor handlerInterceptor;
public final SessionManager sessionManager;
/*基础服务端*/
public Server server;
public String name;
/*服务名*/
public final ServerType type;
public NettyConfig(int workerGroup,
int businessGroup,
int readerIdleTime,
int writerIdleTime,
int allIdleTime,
Integer port,
Integer maxFrameLength,
LengthField lengthField,
Delimiter[] delimiters,
MessageDecoder decoder,
MessageEncoder encoder,
HandlerMapping handlerMapping,
HandlerInterceptor handlerInterceptor,
SessionManager sessionManager,
ServerType type,
String name,
Server server) {
/*校验值是否正确*/
ObjectUtil.checkNotNull(port, FastBeeConstant.SERVER.PORT);
ObjectUtil.checkPositive(port, FastBeeConstant.SERVER.PORT);
if (ServerType.UDP == type || ServerType.TCP == type){
ObjectUtil.checkNotNull(decoder, "decoder");
ObjectUtil.checkNotNull(encoder, "encoder");
ObjectUtil.checkNotNull(handlerMapping, "handlerMapping");
ObjectUtil.checkNotNull(handlerInterceptor, "handlerInterceptor");
}
if (type == ServerType.TCP){
ObjectUtil.checkNotNull(maxFrameLength, FastBeeConstant.SERVER.MAXFRAMELENGTH);
ObjectUtil.checkPositive(maxFrameLength, FastBeeConstant.SERVER.MAXFRAMELENGTH);
// ObjectUtil.checkNotNull(delimiters,FastBeeConstant.SERVER.DELIMITERS);
}
/*获取核数*/
int processors = NettyRuntime.availableProcessors();
this.workerCore = workerGroup > 0 ? workerGroup : processors + 2;
this.businessCore = businessGroup > 0 ? businessGroup : Math.max(1, processors >> 1);
this.readerIdleTime = readerIdleTime;
this.writerIdleTime = writerIdleTime;
this.allIdleTime = allIdleTime;
this.port = port;
this.maxFrameLength = maxFrameLength;
this.lengthField = lengthField;
this.delimiters = delimiters;
this.decoder = decoder;
this.encoder = encoder;
this.handlerMapping = handlerMapping;
this.handlerInterceptor = handlerInterceptor;
this.sessionManager = sessionManager != null ? sessionManager : new SessionManager();
this.type = type;
switch (type){
case MQTT:
case WEBSOCKET:
this.name = name != null ? name : ServerType.MQTT.name();
this.server = server;
this.server.config = this;
break;
default:
}
}
public Server build() {
return server;
}
public static NettyConfig.Builder custom() {
return new Builder();
}
public static class Builder {
private int workerCore;
private int businessCore ;
private int readerIdleTime = 240;
private int writerIdleTime = 0;
private int allIdleTime = 0;
private Integer port;
private Integer maxFrameLength;
private LengthField lengthField;
private Delimiter[] delimiters;
private MessageDecoder decoder;
private MessageEncoder encoder;
private HandlerMapping handlerMapping;
private HandlerInterceptor handlerInterceptor;
private SessionManager sessionManager;
private ServerType type;
private String name;
private Server server;
public Builder() {
}
public Builder setThreadGroup(int workerCore, int businessCore) {
this.workerCore = workerCore;
this.businessCore = businessCore;
return this;
}
public Builder setIdleStateTime(int readerIdleTime, int writerIdleTime, int allIdleTime) {
this.readerIdleTime = readerIdleTime;
this.writerIdleTime = writerIdleTime;
this.allIdleTime = allIdleTime;
return this;
}
public Builder setPort(Integer port) {
this.port = port;
return this;
}
public Builder setServer(Server server){
this.server = server;
return this;
}
public Builder setMaxFrameLength(Integer maxFrameLength) {
this.maxFrameLength = maxFrameLength;
return this;
}
public Builder setLengthField(LengthField lengthField) {
this.lengthField = lengthField;
return this;
}
public Builder setDelimiters(byte[][] delimiters) {
Delimiter[] t = new Delimiter[delimiters.length];
for (int i = 0; i < delimiters.length; i++) {
t[i] = new Delimiter(delimiters[i]);
}
this.delimiters = t;
return this;
}
public Builder setDelimiters(Delimiter... delimiters) {
this.delimiters = delimiters;
return this;
}
public Builder setDecoder(MessageDecoder decoder) {
this.decoder = decoder;
return this;
}
public Builder setEncoder(MessageEncoder encoder) {
this.encoder = encoder;
return this;
}
public Builder setHandlerMapping(HandlerMapping handlerMapping) {
this.handlerMapping = handlerMapping;
return this;
}
public Builder setHandlerInterceptor(HandlerInterceptor handlerInterceptor) {
this.handlerInterceptor = handlerInterceptor;
return this;
}
public Builder setSessionManager(SessionManager sessionManager) {
this.sessionManager = sessionManager;
return this;
}
public Builder setType(ServerType type){
this.type = type;
return this;
}
public Builder setName(String name) {
this.name = name;
return this;
}
public Server build() {
return new NettyConfig(
this.workerCore,
this.businessCore,
this.readerIdleTime,
this.writerIdleTime,
this.allIdleTime,
this.port,
this.maxFrameLength,
this.lengthField,
this.delimiters,
this.decoder,
this.encoder,
this.handlerMapping,
this.handlerInterceptor,
this.sessionManager,
this.type,
this.name,
this.server
).build();
}
}
}

View File

@@ -0,0 +1,28 @@
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<parent>
<artifactId>fastbee-server</artifactId>
<groupId>com.fastbee</groupId>
<version>3.8.5</version>
</parent>
<artifactId>mqtt-broker</artifactId>
<description>基于netty搭建的mqttBroker</description>
<dependencies>
<dependency>
<groupId>com.fastbee</groupId>
<artifactId>iot-server-core</artifactId>
</dependency>
<dependency>
<groupId>com.fastbee</groupId>
<artifactId>fastbee-mq</artifactId>
</dependency>
</dependencies>
</project>

View File

@@ -0,0 +1,21 @@
package com.fastbee.mqtt.annotation;
import io.netty.handler.codec.mqtt.MqttMessageType;
import org.springframework.stereotype.Component;
import java.lang.annotation.ElementType;
import java.lang.annotation.Retention;
import java.lang.annotation.RetentionPolicy;
import java.lang.annotation.Target;
@Target(ElementType.TYPE)
@Retention(RetentionPolicy.RUNTIME)
/*自动注入bean*/
@Component
public @interface Process {
/*消息类型*/
MqttMessageType type() default MqttMessageType.PUBLISH;
}

View File

@@ -0,0 +1,101 @@
package com.fastbee.mqtt.auth;
import com.fastbee.common.constant.Constants;
import com.fastbee.common.constant.FastBeeConstant;
import com.fastbee.common.core.redis.RedisCache;
import com.fastbee.common.exception.ServiceException;
import com.fastbee.common.utils.StringUtils;
import com.fastbee.iot.model.MqttAuthenticationModel;
import com.fastbee.iot.service.IToolService;
import com.fastbee.mq.mqttClient.MqttClientConfig;
import io.jsonwebtoken.Claims;
import io.jsonwebtoken.Jwts;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.http.HttpStatus;
import org.springframework.http.ResponseEntity;
import org.springframework.stereotype.Component;
import javax.annotation.Resource;
/**
* 客户端认证
*
* @author gsb
* @date 2022/9/15 13:40
*/
@Slf4j
@Component
public class AuthService {
@Resource
private IToolService toolService;
@Resource
private RedisCache redisCache;
@Resource
private MqttClientConfig mqttConfig;
// 令牌秘钥
@Value("${token.secret}")
private String secret;
@Value("${server.broker.must-pass}")
private boolean mustPass;
/**
* MQTT客户端认证
*
* @param clientId 客户端id
* @param username 用户名
* @param password 密码
* @return 结果
*/
public boolean auth(String clientId, String username, String password) {
//不需要账号密码校验,直接返回true
if (!mustPass) return true;
if (StringUtils.isEmpty(clientId) || StringUtils.isEmpty(username) || StringUtils.isEmpty(password)) {
log.error("=>客户端参数缺少,clientId:{},username:{},password:{}", clientId, username, password);
return false;
}
try {
if (clientId.startsWith("server")) {
// 服务端认证:配置的账号密码认证
if (mqttConfig.getUsername().equals(username) && mqttConfig.getPassword().equals(password)) {
log.info("-----------服务端mqtt认证成功,clientId:" + clientId + "---------------");
return true;
} else {
ResponseEntity response = toolService.returnUnauthorized(new MqttAuthenticationModel(clientId, username, password), "mqtt账号和密码与认证服务器配置不匹配");
log.warn("=>服务端认证失败[{}]", response.getBody());
throw new ServiceException("服务端认证失败:"+response.getBody());
}
} else if (clientId.startsWith("web") || clientId.startsWith("phone")) {
// web端和移动端认证token认证
String token = password;
if (StringUtils.isNotEmpty(token) && token.startsWith(Constants.TOKEN_PREFIX)) {
token = token.replace(Constants.TOKEN_PREFIX, "");
}
try {
Claims claims = Jwts.parser().setSigningKey(secret).parseClaimsJws(token).getBody();
log.info("-----------移动端/Web端mqtt认证成功,clientId:" + clientId + "---------------");
return true;
} catch (Exception ex) {
ResponseEntity response = toolService.returnUnauthorized(new MqttAuthenticationModel(clientId, username, password), ex.getMessage());
log.warn("=>移动端/Web端mqtt认证失败[{}]",response.getBody());
throw new ServiceException("移动端/Web端mqtt认证失败:"+response.getBody());
}
} else {
// 设备端认证
ResponseEntity response = toolService.clientAuth(clientId, username, password);
if (response.getStatusCodeValue() == HttpStatus.OK.value()) {
return true;
} else {
log.warn("=>设备端认证失败");
throw new ServiceException("设备端认证失败:"+response.getBody());
}
}
} catch (Exception e) {
log.error("=>客户端认证失败", e);
return false;
}
}
}

View File

@@ -0,0 +1,30 @@
package com.fastbee.mqtt.codec;
import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.MessageToMessageCodec;
import io.netty.handler.codec.http.websocketx.BinaryWebSocketFrame;
import org.springframework.stereotype.Component;
import java.util.List;
/**
* ws协议与MQTT协议编码转换
* @author gsb
* @date 2022/9/15 14:35
*/
@ChannelHandler.Sharable
@Component
public class WebSocketMqttCodec extends MessageToMessageCodec<BinaryWebSocketFrame, ByteBuf> {
@Override
protected void encode(ChannelHandlerContext channelHandlerContext, ByteBuf byteBuf, List<Object> list) throws Exception {
list.add(new BinaryWebSocketFrame(byteBuf.retain()));
}
@Override
protected void decode(ChannelHandlerContext channelHandlerContext, BinaryWebSocketFrame binaryWebSocketFrame, List<Object> list) throws Exception {
list.add(binaryWebSocketFrame.retain().content());
}
}

View File

@@ -0,0 +1,137 @@
package com.fastbee.mqtt.handler;
import com.fastbee.common.constant.FastBeeConstant;
import com.fastbee.common.core.redis.RedisCache;
import com.fastbee.common.enums.ServerType;
import com.fastbee.common.utils.DateUtils;
import com.fastbee.mqtt.annotation.Process;
import com.fastbee.mqtt.auth.AuthService;
import com.fastbee.mqtt.handler.adapter.MqttHandler;
import com.fastbee.mqtt.manager.ResponseManager;
import com.fastbee.mqtt.manager.SessionManger;
import com.fastbee.mqtt.manager.WillMessageManager;
import com.fastbee.mqtt.model.WillMessage;
import com.fastbee.base.session.Session;
import com.fastbee.base.util.AttributeUtils;
import com.fastbee.mqtt.utils.MqttMessageUtils;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.mqtt.*;
import io.netty.util.CharsetUtil;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import javax.annotation.Resource;
import java.net.InetSocketAddress;
@Slf4j
@Process(type = MqttMessageType.CONNECT)
public class MqttConnect implements MqttHandler {
@Autowired
private AuthService authService;
@Resource
private RedisCache redisCache;
@Override
public void handler(ChannelHandlerContext ctx, MqttMessage message) {
MqttConnectMessage connectMessage = (MqttConnectMessage) message;
/*获取客户端Id*/
String clientId = connectMessage.payload().clientIdentifier();
if (clientId.contains("&")){
clientId = clientId.split("&")[1];
}
log.debug("=>客户端:{} 连接:{}", clientId, message);
/*获取session*/
Session session = new Session();
/*mqtt版本*/
MqttVersion version = MqttVersion.fromProtocolNameAndLevel(connectMessage.variableHeader().name(), (byte) connectMessage.variableHeader().version());
boolean cleanSession = connectMessage.variableHeader().isCleanSession();
session.setHandlerContext(ctx);
session.setVersion(version);
session.setClientId(clientId);
session.setCleanSession(cleanSession);
session.setUsername(connectMessage.payload().userName());
session.setConnected_at(DateUtils.getNowDate());
InetSocketAddress socketAddress = (InetSocketAddress) ctx.channel().remoteAddress();
session.setIp(socketAddress.getAddress().getHostAddress());
session.setServerType(ServerType.MQTT);
/*设置客户端ping时间*/
MqttConnectVariableHeader header = connectMessage.variableHeader();
if (header.keepAliveTimeSeconds() > 0 && session.getKeepAliveMax() >= header.keepAliveTimeSeconds()) {
session.setKeepAlive(header.keepAliveTimeSeconds());
}
/*mqtt客户端登录校验*/
if (!check(session, connectMessage)) {
log.error("=>客户端:{},连接异常", clientId);
session.getHandlerContext().close();
return;
}
/*保存ClientId 和 session 到Attribute*/
AttributeUtils.setClientId(ctx.channel(), clientId);
AttributeUtils.setSession(ctx.channel(), session);
SessionManger.removeClient(clientId);
session.setConnected(true);
SessionManger.buildSession(clientId, session);
handleWill(connectMessage);
ResponseManager.responseMessage(session,
MqttMessageFactory.newMessage(
new MqttFixedHeader(MqttMessageType.CONNACK, false, MqttQoS.AT_MOST_ONCE, false, 0x02),
new MqttConnAckVariableHeader(MqttConnectReturnCode.CONNECTION_ACCEPTED, false),
null),
true
);
}
/**
* 遗嘱消息处理
*
* @param message 连接消息
*/
private void handleWill(MqttConnectMessage message) {
/*如果没有设置处理遗嘱消息,返回*/
if (!message.variableHeader().isWillFlag()) {
return;
}
/*生成客户端model*/
MqttPublishMessage publishMessage = (MqttPublishMessage) MqttMessageFactory.newMessage(
new MqttFixedHeader(MqttMessageType.PUBLISH, false,
MqttQoS.valueOf(message.variableHeader().willQos()),
message.variableHeader().isWillRetain(), 0),
new MqttPublishVariableHeader(message.payload().willTopic(), 0),
Unpooled.buffer().writeBytes(message.payload().willMessageInBytes()));
WillMessage msg = new WillMessage(message.payload().clientIdentifier(),
message.variableHeader().isCleanSession(), message.payload().willTopic(), publishMessage);
WillMessageManager.push(msg);
}
/**
* 客户端连接校验
*
* @param session 客户端
* @param message 连接消息
* @return 结果
*/
private boolean check(Session session, MqttConnectMessage message) {
/*获取客户端连接地址*/
InetSocketAddress address = (InetSocketAddress) session.getHandlerContext().channel().remoteAddress();
String host = address.getAddress().getHostAddress();
/*webSocket客户端 系统内部客户端不校验*/
String clientId = message.payload().clientIdentifier();
/*根据用户名,密码校验*/
String username = message.payload().userName();
String password = message.payload().passwordInBytes() == null ? null : new String(message.payload().passwordInBytes(), CharsetUtil.UTF_8);
/*验证失败,应答客户端*/
if (!authService.auth(clientId, username, password)) {
MqttConnAckMessage connAckMessage = MqttMessageUtils.buildConntAckMessage(MqttConnectReturnCode.CONNECTION_REFUSED_BAD_USER_NAME_OR_PASSWORD, false);
ResponseManager.responseMessage(session, connAckMessage, true);
return false;
}
return true;
}
}

View File

@@ -0,0 +1,41 @@
package com.fastbee.mqtt.handler;
import com.fastbee.mqtt.annotation.Process;
import com.fastbee.mqtt.handler.adapter.MqttHandler;
import com.fastbee.mqtt.manager.ClientManager;
import com.fastbee.mqtt.manager.SessionManger;
import com.fastbee.base.session.Session;
import com.fastbee.base.util.AttributeUtils;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.mqtt.MqttMessage;
import io.netty.handler.codec.mqtt.MqttMessageType;
import lombok.extern.slf4j.Slf4j;
@Process(type = MqttMessageType.DISCONNECT)
@Slf4j
public class MqttDisConnect implements MqttHandler {
@Override
public void handler(ChannelHandlerContext ctx, MqttMessage message) {
/*获取clientId*/
String clientId = AttributeUtils.getClientId(ctx.channel());
/*获取session*/
Session session = AttributeUtils.getSession(ctx.channel());
log.debug("=>客户端正常断开,clientId:[{}]", clientId);
try {
if (!session.getConnected()) {
session.getHandlerContext().close();
return;
}
/*处理断开客户端连接*/
SessionManger.pingTimeout(session.getClientId());
/*移除相关topic*/
ClientManager.remove(session.getClientId());
} catch (Exception e) {
log.error("=>客户端断开连接异常:{}", session);
}
}
}

View File

@@ -0,0 +1,37 @@
package com.fastbee.mqtt.handler;
import com.fastbee.mqtt.annotation.Process;
import com.fastbee.mqtt.handler.adapter.MqttHandler;
import com.fastbee.mqtt.manager.ClientManager;
import com.fastbee.mqtt.manager.ResponseManager;
import com.fastbee.base.util.AttributeUtils;
import com.fastbee.mqtt.utils.MqttMessageUtils;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.mqtt.*;
import lombok.extern.slf4j.Slf4j;
/**
* 客户端Ping消息应答
*
* @author bill
*/
@Slf4j
@Process(type = MqttMessageType.PINGREQ)
public class MqttPingreq implements MqttHandler {
@Override
public void handler(ChannelHandlerContext ctx, MqttMessage message) {
/*获取客户端id*/
String clientId = AttributeUtils.getClientId(ctx.channel());
try {
// log.debug("=>客户端:{},心跳信息", clientId);
/*更新客户端ping时间*/
ClientManager.updatePing(clientId);
/*响应设备的ping消息*/
MqttMessage pingResp = MqttMessageUtils.buildPingResp();
ResponseManager.sendMessage(pingResp, clientId, true);
} catch (Exception e) {
log.error("=>客户端:{},ping异常:{}", clientId, e);
}
}
}

View File

@@ -0,0 +1,39 @@
package com.fastbee.mqtt.handler;
import com.fastbee.mqtt.annotation.Process;
import com.fastbee.mqtt.handler.adapter.MqttHandler;
import com.fastbee.mqtt.manager.ClientManager;
import com.fastbee.mqtt.service.IMessageStore;
import com.fastbee.base.session.Session;
import com.fastbee.base.util.AttributeUtils;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.mqtt.MqttMessage;
import io.netty.handler.codec.mqtt.MqttMessageType;
import io.netty.handler.codec.mqtt.MqttPubAckMessage;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
/**
* 消息等级=Qos1,收到发布消息确认
*
* @author bill
*/
@Process(type = MqttMessageType.PUBACK)
@Slf4j
public class MqttPubAck implements MqttHandler {
@Autowired
private IMessageStore messageStore;
@Override
public void handler(ChannelHandlerContext ctx, MqttMessage message) {
MqttPubAckMessage ackMessage = (MqttPubAckMessage) message;
// PacketId
int packetId = ackMessage.variableHeader().messageId();
Session session = AttributeUtils.getSession(ctx.channel());
// Qos1 的存储消息释放
messageStore.removePubMsg(packetId);
/*更新平台ping*/
ClientManager.updatePing(session.getClientId());
}
}

View File

@@ -0,0 +1,47 @@
package com.fastbee.mqtt.handler;
import com.fastbee.mqtt.annotation.Process;
import com.fastbee.mqtt.handler.adapter.MqttHandler;
import com.fastbee.mqtt.manager.ClientManager;
import com.fastbee.mqtt.manager.ResponseManager;
import com.fastbee.mqtt.service.IMessageStore;
import com.fastbee.base.session.Session;
import com.fastbee.base.util.AttributeUtils;
import com.fastbee.mqtt.utils.MqttMessageUtils;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.mqtt.MqttMessage;
import io.netty.handler.codec.mqtt.MqttMessageIdVariableHeader;
import io.netty.handler.codec.mqtt.MqttMessageType;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
/**
* 消息等级=Qos2 发布消息收到 交付第一步
*
* @author bill
*/
@Process(type = MqttMessageType.PUBREC)
@Slf4j
public class MqttPubRec implements MqttHandler {
@Autowired
private IMessageStore messageStore;
@Override
public void handler(ChannelHandlerContext ctx, MqttMessage message) {
MqttMessageIdVariableHeader variableHeader = MqttMessageUtils.getIdVariableHeader(message);
//clientId
String clientId = AttributeUtils.getClientId(ctx.channel());
Session session = AttributeUtils.getSession(ctx.channel());
//获取packetId
int messageId = variableHeader.messageId();
/*释放消息*/
messageStore.removePubMsg(messageId);
messageStore.saveRelOutMsg(messageId);
// 回复REL 进入第二阶段
MqttMessage mqttMessage = MqttMessageUtils.buildPubRelMessage(message);
ResponseManager.responseMessage(session, mqttMessage, true);
/*更新平台ping*/
ClientManager.updatePing(session.getClientId());
}
}

View File

@@ -0,0 +1,41 @@
package com.fastbee.mqtt.handler;
import com.fastbee.mqtt.annotation.Process;
import com.fastbee.mqtt.handler.adapter.MqttHandler;
import com.fastbee.mqtt.manager.ClientManager;
import com.fastbee.mqtt.manager.ResponseManager;
import com.fastbee.mqtt.service.IMessageStore;
import com.fastbee.base.session.Session;
import com.fastbee.base.util.AttributeUtils;
import com.fastbee.mqtt.utils.MqttMessageUtils;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.mqtt.MqttMessage;
import io.netty.handler.codec.mqtt.MqttMessageIdVariableHeader;
import io.netty.handler.codec.mqtt.MqttMessageType;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
/**
* 消息等级=Qos2 发布消息释放 PUBREL
* @author bill
*/
@Slf4j
@Process(type = MqttMessageType.PUBREL)
public class MqttPubRel implements MqttHandler {
@Autowired
private IMessageStore messageStore;
@Override
public void handler(ChannelHandlerContext ctx, MqttMessage message){
MqttMessageIdVariableHeader variableHeader = MqttMessageUtils.getIdVariableHeader(message);
Session session = AttributeUtils.getSession(ctx.channel());
//获取packetId
int messageId = variableHeader.messageId();
messageStore.removeRelInMsg(messageId);
MqttMessage mqttMessage = MqttMessageUtils.buildPubCompMessage(message);
ResponseManager.responseMessage(session,mqttMessage,true);
/*更新本地ping时间*/
ClientManager.updatePing(session.getClientId());
}
}

View File

@@ -0,0 +1,38 @@
package com.fastbee.mqtt.handler;
import com.fastbee.mqtt.annotation.Process;
import com.fastbee.mqtt.handler.adapter.MqttHandler;
import com.fastbee.mqtt.manager.ClientManager;
import com.fastbee.mqtt.service.IMessageStore;
import com.fastbee.base.session.Session;
import com.fastbee.base.util.AttributeUtils;
import com.fastbee.mqtt.utils.MqttMessageUtils;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.mqtt.MqttMessage;
import io.netty.handler.codec.mqtt.MqttMessageIdVariableHeader;
import io.netty.handler.codec.mqtt.MqttMessageType;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
/**
* 消息等级=Qos2 发布消息完成
* @author bill
*/
@Process(type = MqttMessageType.PUBCOMP)
@Slf4j
public class MqttPubcomp implements MqttHandler {
@Autowired
private IMessageStore messageStore;
@Override
public void handler(ChannelHandlerContext ctx, MqttMessage message){
MqttMessageIdVariableHeader variableHeader = MqttMessageUtils.getIdVariableHeader(message);
Session session = AttributeUtils.getSession(ctx.channel());
//获取packetId
int messageId = variableHeader.messageId();
messageStore.removeRelOutMsg(messageId);
/*更新平台ping*/
ClientManager.updatePing(session.getClientId());
}
}

View File

@@ -0,0 +1,204 @@
package com.fastbee.mqtt.handler;
import com.fastbee.common.constant.FastBeeConstant;
import com.fastbee.common.core.mq.DeviceReportBo;
import com.fastbee.common.core.redis.RedisCache;
import com.fastbee.common.enums.ServerType;
import com.fastbee.common.utils.DateUtils;
import com.fastbee.common.utils.gateway.mq.TopicsUtils;
import com.fastbee.mq.redischannel.producer.MessageProducer;
import com.fastbee.mq.service.IDeviceReportMessageService;
import com.fastbee.mqtt.annotation.Process;
import com.fastbee.mqtt.handler.adapter.MqttHandler;
import com.fastbee.mqtt.manager.ClientManager;
import com.fastbee.mqtt.manager.ResponseManager;
import com.fastbee.mqtt.manager.RetainMsgManager;
import com.fastbee.mqtt.model.ClientMessage;
import com.fastbee.mqtt.service.IMessageStore;
import com.fastbee.base.session.Session;
import com.fastbee.base.util.AttributeUtils;
import io.netty.buffer.ByteBufUtil;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.mqtt.*;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import javax.annotation.Resource;
/**
* 客户端消息推送处理类
*
* @author bill
*/
@Slf4j
@Process(type = MqttMessageType.PUBLISH)
public class MqttPublish implements MqttHandler {
@Autowired
private IMessageStore messageStore;
@Resource
private TopicsUtils topicsUtils;
@Resource
private RedisCache redisCache;
@Resource
private IDeviceReportMessageService deviceReportMessageService;
@Override
public void handler(ChannelHandlerContext ctx, MqttMessage message) {
MqttPublishMessage publishMessage = (MqttPublishMessage) message;
/*获取客户端id*/
String clientId = AttributeUtils.getClientId(ctx.channel());
String topicName = publishMessage.variableHeader().topicName();
log.debug("=>***客户端[{}],主题[{}],推送消息[{}]", clientId, topicName,
ByteBufUtil.hexDump(publishMessage.content()));
// 以get结尾是模拟客户端数据,只转发消息
if (topicName.endsWith(FastBeeConstant.MQTT.PROPERTY_GET_SIMULATE)) {
sendTestToMQ(publishMessage);
} else {
/*获取客户端session*/
Session session = AttributeUtils.getSession(ctx.channel());
/*推送保留信息*/
pubRetain(publishMessage);
/*响应客户端消息到达Broker*/
callBack(session, publishMessage, clientId);
/*推送到订阅的客户端*/
sendMessageToClients(publishMessage);
/*推送到MQ处理*/
sendToMQ(publishMessage);
/*累计接收消息数*/
redisCache.incr2(FastBeeConstant.REDIS.MESSAGE_RECEIVE_TOTAL, -1L);
redisCache.incr2(FastBeeConstant.REDIS.MESSAGE_RECEIVE_TODAY, 60 * 60 * 24);
}
}
/**
* 消息推送
*
* @param message 推送消息
*/
@SneakyThrows
public void sendToMQ(MqttPublishMessage message) {
/*获取topic*/
String topicName = message.variableHeader().topicName();
/*只处理上报数据*/
if (!topicName.endsWith(FastBeeConstant.MQTT.UP_TOPIC_SUFFIX)) {
return;
}
DeviceReportBo reportBo = DeviceReportBo.builder()
.serialNumber(topicsUtils.parseSerialNumber(topicName))
.topicName(topicName)
.packetId((long) message.variableHeader().packetId())
.platformDate(DateUtils.getNowDate())
.data(ByteBufUtil.getBytes(message.content()))
.serverType(ServerType.MQTT)
.build();
if (topicName.endsWith(FastBeeConstant.TOPIC.MSG_REPLY) ||
topicName.endsWith(FastBeeConstant.TOPIC.SUB_UPGRADE_REPLY) ||
topicName.endsWith(FastBeeConstant.TOPIC.UPGRADE_REPLY)) {
/*设备应答服务器回调数据*/
reportBo.setReportType(2);
} else {
/*设备上报数据*/
reportBo.setReportType(1);
}
if (topicName.contains("property")) {
deviceReportMessageService.parseReportMsg(reportBo);
}
}
/**
* 发送模拟数据进行处理
* @param message
*/
public void sendTestToMQ(MqttPublishMessage message) {
/*获取topic*/
String topicName = message.variableHeader().topicName();
DeviceReportBo reportBo = DeviceReportBo.builder()
.serialNumber(topicsUtils.parseSerialNumber(topicName))
.topicName(topicName)
.packetId((long) message.variableHeader().packetId())
.platformDate(DateUtils.getNowDate())
.data(ByteBufUtil.getBytes(message.content()))
.build();
MessageProducer.sendOtherMsg(reportBo);
}
/**
* 推送消息到订阅客户端
*
* @param message 消息
*/
public void sendMessageToClients(MqttPublishMessage message) {
ClientManager.pubTopic(message);
}
/**
* 应答客户端消息到达Broker
*
* @param session 客户端
* @param message 消息
*/
private void callBack(Session session, MqttPublishMessage message, String clientId) {
/*获取消息等级*/
MqttQoS mqttQoS = message.fixedHeader().qosLevel();
int packetId = message.variableHeader().packetId();
MqttFixedHeader header;
switch (mqttQoS.value()) {
/*0,1消息等级直接回复*/
case 0:
case 1:
header = new MqttFixedHeader(MqttMessageType.PUBACK, false, mqttQoS, false, 0);
break;
case 2:
// 处理Qos2的消息确认
if (!messageStore.outRelContains(packetId)) {
messageStore.saveRelInMsg(packetId);
}
header = new MqttFixedHeader(MqttMessageType.PUBREC, false, MqttQoS.AT_MOST_ONCE, false, 0);
break;
default:
header = null;
}
/*处理消息等级*/
handleMqttQos(packetId, mqttQoS, true, clientId);
/*响应客户端*/
MqttMessageIdVariableHeader variableHeader = null;
if (packetId > 0) {
variableHeader = MqttMessageIdVariableHeader.from(packetId);
}
MqttPubAckMessage ackMessage = new MqttPubAckMessage(header, variableHeader);
if (mqttQoS.value() >= 1) {
ResponseManager.responseMessage(session, ackMessage, true);
}
/*更新客户端ping时间*/
ClientManager.updatePing(session.getClientId());
}
/**
* Qos不同消息处理
*/
private void handleMqttQos(int packetId, MqttQoS qoS, boolean clearSession, String clientId) {
if (qoS == MqttQoS.AT_LEAST_ONCE || qoS == MqttQoS.EXACTLY_ONCE) {
ClientMessage clientMessage = ClientMessage.of(clientId, qoS, null, false);
messageStore.savePubMsg(packetId, clientMessage);
}
}
/**
* 推送保留信息
*/
@SneakyThrows
private void pubRetain(MqttPublishMessage message) {
redisCache.incr2(FastBeeConstant.REDIS.MESSAGE_RETAIN_TOTAL, -1L);
/*根据message.fixedHeader().isRetain() 判断是否有保留信息*/
RetainMsgManager.pushMessage(message);
}
}

View File

@@ -0,0 +1,107 @@
package com.fastbee.mqtt.handler;
import com.alibaba.fastjson2.JSON;
import com.fastbee.common.constant.FastBeeConstant;
import com.fastbee.common.core.redis.RedisCache;
import com.fastbee.mqtt.annotation.Process;
import com.fastbee.mqtt.model.ClientMessage;
import com.fastbee.base.session.Session;
import com.fastbee.base.util.AttributeUtils;
import com.fastbee.mqtt.utils.MqttMessageUtils;
import com.fastbee.mqtt.model.RetainMessage;
import com.fastbee.mqtt.handler.adapter.MqttHandler;
import com.fastbee.mqtt.manager.ClientManager;
import com.fastbee.mqtt.manager.ResponseManager;
import com.fastbee.mqtt.manager.RetainMsgManager;
import com.fastbee.common.utils.gateway.mq.TopicsUtils;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.mqtt.*;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import javax.annotation.Resource;
import java.util.List;
import java.util.stream.Collectors;
@Slf4j
@Process(type = MqttMessageType.SUBSCRIBE)
public class MqttSubscribe implements MqttHandler {
@Resource
private RedisCache redisCache;
@Override
public void handler(ChannelHandlerContext ctx, MqttMessage message) {
subscribe(ctx, (MqttSubscribeMessage) message);
}
public void subscribe(ChannelHandlerContext ctx, MqttSubscribeMessage message) {
/*获取session*/
Session session = AttributeUtils.getSession(ctx.channel());
/*获取客户端订阅的topic列表*/
List<MqttTopicSubscription> topList = message.payload().topicSubscriptions();
/*获取topicName列表*/
List<String> topicNameList = topList.stream().map(MqttTopicSubscription::topicName).collect(Collectors.toList());
log.debug("=>客户端:{},订阅主题:{}", session.getClientId(), JSON.toJSONString(topicNameList));
if (!TopicsUtils.validTopicFilter(topicNameList)) {
log.error("=>订阅主题不合法:{}", JSON.toJSONString(topicNameList));
return;
}
/*存储到本地topic缓存*/
topicNameList.forEach(topicName -> {
ClientManager.push(topicName, session);
/*累计订阅数*/
redisCache.incr2(FastBeeConstant.REDIS.MESSAGE_SUBSCRIBE_TOTAL,-1L);
});
/*更新客户端ping*/
ClientManager.updatePing(session.getClientId());
/*应答客户端订阅成功*/
MqttSubAckMessage subAckMessage = MqttMessageUtils.buildSubAckMessage(message);
ResponseManager.responseMessage(session, subAckMessage, true);
/*客户端订阅了遗留消息主题后,推送遗留消息给客户端*/
topList.forEach(topic -> {
retain(topic.topicName(), session, topic.qualityOfService());
});
}
/**
* 推送遗留消息
*
* @param topicName 主题
* @param session 客户端
* @param mqttQoS 消息质量
*/
@SneakyThrows
private void retain(String topicName, Session session, MqttQoS mqttQoS) {
RetainMessage message = RetainMsgManager.getRetain(topicName);
if (null == message) {
return;
}
MqttQoS qos = message.getQos() > mqttQoS.value() ? mqttQoS : MqttQoS.valueOf(message.getQos());
switch (qos.value()) {
case 0:
buildMessage(qos, topicName, 0, message.getMessage(), session);
break;
case 1:
case 2:
/*使用实时时间戳充当 packId*/
buildMessage(qos, topicName, (int) System.currentTimeMillis(), message.getMessage(), session);
break;
}
}
/*组装推送数据*/
private void buildMessage(MqttQoS qos, String topicName, int packetId, byte[] message, Session session) {
/*生成客户端model*/
ClientMessage clientMessage = ClientMessage.of(qos, topicName, false, message);
/*组建推送消息*/
MqttPublishMessage publishMessage = MqttMessageUtils.buildPublishMessage(clientMessage, packetId);
/*推送消息*/
ResponseManager.publishClients(publishMessage, session);
}
}

View File

@@ -0,0 +1,38 @@
package com.fastbee.mqtt.handler;
import com.fastbee.mqtt.annotation.Process;
import com.fastbee.mqtt.handler.adapter.MqttHandler;
import com.fastbee.mqtt.manager.ClientManager;
import com.fastbee.mqtt.manager.ResponseManager;
import com.fastbee.base.session.Session;
import com.fastbee.base.util.AttributeUtils;
import com.fastbee.mqtt.utils.MqttMessageUtils;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.mqtt.MqttMessage;
import io.netty.handler.codec.mqtt.MqttMessageType;
import io.netty.handler.codec.mqtt.MqttUnsubAckMessage;
import io.netty.handler.codec.mqtt.MqttUnsubscribeMessage;
import lombok.extern.slf4j.Slf4j;
import java.util.List;
@Slf4j
@Process(type = MqttMessageType.UNSUBSCRIBE)
public class MqttUnsubscribe implements MqttHandler {
@Override
public void handler(ChannelHandlerContext ctx, MqttMessage message) {
MqttUnsubscribeMessage unsubscribeMessage = (MqttUnsubscribeMessage) message;
List<String> topics = unsubscribeMessage.payload().topics();
log.debug("=>收到取消订阅请求,topics[{}]", topics);
Session session = AttributeUtils.getSession(ctx.channel());
topics.forEach(topic -> {
ClientManager.unsubscribe(topic, session);
});
MqttUnsubAckMessage unsubAckMessage = MqttMessageUtils.buildUnsubAckMessage(unsubscribeMessage);
ResponseManager.responseMessage(session, unsubAckMessage, true);
/*更新客户端平台时间*/
ClientManager.updatePing(session.getClientId());
}
}

View File

@@ -0,0 +1,12 @@
package com.fastbee.mqtt.handler.adapter;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.mqtt.MqttMessage;
/**
* @author bill
*/
public interface MqttHandler {
public void handler(ChannelHandlerContext ctx, MqttMessage message);
}

View File

@@ -0,0 +1,89 @@
package com.fastbee.mqtt.handler.adapter;
import com.fastbee.common.exception.iot.MqttAuthorizationException;
import com.fastbee.common.exception.iot.MqttClientUserNameOrPassException;
import com.fastbee.mqtt.manager.SessionManger;
import com.fastbee.base.util.AttributeUtils;
import com.fastbee.mqtt.utils.MqttMessageUtils;
import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.handler.codec.mqtt.*;
import io.netty.handler.timeout.IdleState;
import io.netty.handler.timeout.IdleStateEvent;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;
import java.io.IOException;
import java.net.InetSocketAddress;
import java.util.Optional;
/**
* @author gsb
* @date 2022/9/15 10:34
*/
@Slf4j
@Component
@ChannelHandler.Sharable
public class MqttMessageAdapter extends SimpleChannelInboundHandler<MqttMessage> {
// @Autowired
private MqttMessageDelegate messageDelegate;
public MqttMessageAdapter(MqttMessageDelegate delegate) {
this.messageDelegate = delegate;
}
/**
* 客户端上报消息处理
*
* @param context 上下文
* @param message 消息
*/
@Override
@SuppressWarnings("ThrowableResultOfMethodCallIgnored")
protected void channelRead0(ChannelHandlerContext context, MqttMessage message) {
try {
/*校验消息*/
if (message.decoderResult().isFailure()) {
exceptionCaught(context, message.decoderResult().cause());
return;
}
messageDelegate.process(context, message);
}catch (Exception e){
log.error("=>数据进栈异常",e);
}
}
/**
* 客户端心跳处理
*/
@Override
public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
InetSocketAddress socketAddress = (InetSocketAddress) ctx.channel().remoteAddress();
String host = socketAddress.getAddress().getHostAddress();
int port = socketAddress.getPort();
String clientId = AttributeUtils.getClientId(ctx.channel());
if (evt instanceof IdleStateEvent) {
IdleStateEvent idleStateEvent = (IdleStateEvent) evt;
IdleState state = idleStateEvent.state();
if (state == IdleState.ALL_IDLE || state == IdleState.READER_IDLE || state == IdleState.WRITER_IDLE) {
log.error("客户端id[{}] 客户端[{}]port:[{}]心跳超时!",clientId, host, port);
/*关闭通道*/
ctx.close();
}
} else {
super.userEventTriggered(ctx, evt);
}
}
/**
* 处理消息异常情况
*/
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
log.error("=>mqtt连接异常",cause);
}
}

View File

@@ -0,0 +1,58 @@
package com.fastbee.mqtt.handler.adapter;
import com.fastbee.common.exception.ServiceException;
import com.fastbee.mqtt.annotation.Process;
import com.fastbee.base.util.AttributeUtils;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.mqtt.MqttMessage;
import io.netty.handler.codec.mqtt.MqttMessageType;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;
import org.springframework.util.CollectionUtils;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Optional;
/**
* 消息代理类,根据注解{@link com.fastbee.mqtt.annotation.Process} 分发处理类
* @author gsb
* @date 2022/10/8 11:50
*/
@Component
@Slf4j
public class MqttMessageDelegate {
/**mqtt报文类型为key报文处理类为值*/
private final Map<MqttMessageType, MqttHandler> processor = new HashMap<>();
public MqttMessageDelegate(List<MqttHandler> handlers){
if (CollectionUtils.isEmpty(handlers)){
throw new ServiceException("报文处理类为空");
}
/*将处理类缓存到map*/
handlers.forEach(handler ->{
Process annotation = handler.getClass().getAnnotation(Process.class);
Optional.ofNullable(annotation)
.map(Process::type)
.ifPresent(messageType ->processor.put(messageType,handler));
});
}
/**
* 匹配报文处理类
*/
public void process(ChannelHandlerContext ctx, MqttMessage message){
/*获取固定头的报文类型*/
MqttMessageType messageType = message.fixedHeader().messageType();
/*处理客户端连接时先判断Attribute是否存储Session*/
if (MqttMessageType.CONNECT != messageType &&
AttributeUtils.getSession(ctx.channel()) == null){
log.error("=>客户端未连接");
throw new ServiceException("客户端未连接");
}
Optional.of(processor.get(messageType))
.ifPresent(mqttHandler -> mqttHandler.handler(ctx,message));
}
}

View File

@@ -0,0 +1,174 @@
package com.fastbee.mqtt.manager;
import com.fastbee.common.constant.FastBeeConstant;
import com.fastbee.common.utils.gateway.mq.TopicsUtils;
import com.fastbee.common.utils.DateUtils;
import com.fastbee.common.utils.StringUtils;
import com.fastbee.base.session.Session;
import io.netty.handler.codec.mqtt.MqttPublishMessage;
import lombok.extern.slf4j.Slf4j;
import org.springframework.util.CollectionUtils;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
/**
* 客户端管理
*
* @author gsb
* @date 2022/9/15 16:02
*/
@Slf4j
public class ClientManager {
/*topic本地缓存*/
public static Map<String, Map<String, Session>> topicMap = new ConcurrentHashMap<>();
/*客户端最后一次ping时间设备不正常断开判断*/
private static Map<String, Long> pingMap = new ConcurrentHashMap<>();
/*客户端与topic关联本地缓存*/
public static Map<String, Map<String, Boolean>> clientTopicMap = new ConcurrentHashMap<>();
/**
* 将client的上下文相关信息添加到映射关系表中
* @param topic 主题
* @param session
*/
public static void push(String topic, Session session) {
try {
/*处理topic对应的topic*/
Map<String, Session> clientMap = topicMap.get(topic);
if (StringUtils.isEmpty(clientMap)) {
clientMap = new ConcurrentHashMap<>();
}
clientMap.put(session.getClientId(), session);
topicMap.put(topic, clientMap);
/*处理client对应的所有topic*/
Map<String, Boolean> topicsMap = null;
if (clientTopicMap.containsKey(session.getClientId())) {
topicsMap = clientTopicMap.get(session.getClientId());
if (!topicsMap.containsKey(topic)) {
topicsMap.put(topic, true);
}
} else {
topicsMap = new HashMap<>();
topicsMap.put(topic, true);
clientTopicMap.put(session.getClientId(), topicsMap);
}
} catch (Exception e) {
log.error("=>clientId映射topic出现异常:{},e=", e.getMessage(), e);
}
}
/**
* 清理对应client下的所有数据
*
* @param clientId 客户端id
*/
public static void remove(String clientId) {
try {
/*移除client对应的topic*/
Map<String, Boolean> topics = clientTopicMap.get(clientId);
if (null != topics) {
/*从topic中移除client*/
for (String key : topics.keySet()) {
Map<String, Session> clientMap = topicMap.get(key);
if (CollectionUtils.isEmpty(clientMap)) {
continue;
}
clientMap.remove(clientId);
}
clientTopicMap.remove(clientId);
}
pingMap.remove(clientId);
} catch (Exception e) {
log.warn("=>移除client[{}]异常", e.getMessage());
}
}
/**
* 客户端取消订阅
* 删除指定topic下的指定client
*
* @param topic 主题
* @param session 客户端
*/
public static void unsubscribe(String topic, Session session) {
try {
Map<String, Session> clientMap = topicMap.get(topic);
if (StringUtils.isEmpty(clientMap)) {
return;
}
Session s = clientMap.get(session.getClientId());
if (null == s) {
return;
}
clientMap.remove(session.getClientId());
} catch (Exception e) {
log.error("=>客户端取消订阅异常:{}", e.getMessage());
}
}
/**
* 将消息发送到指定topic下的所有client上去
*
* @param msg 推送消息
*/
public static void pubTopic(MqttPublishMessage msg) {
String topic = msg.variableHeader().topicName();
List<String> topicList = TopicsUtils.searchTopic(topic);
for (String itemTopic : topicList) {
Map<String, Session> clientMap = topicMap.get(itemTopic);
if (StringUtils.isEmpty(clientMap)) {
continue;
}
for (Session session : clientMap.values()) {
String clientId = session.getClientId();
if (!validClient(clientId)) {
///*ws的客户端不正常断开连接后直接移除所有信息*/
//if (session.getClientId().startsWith(FastBeeConstant.SERVER.WS_PREFIX)) {
// log.debug("=>移除ws客户端,clientId={}", session);
// remove(clientId);
//}
log.warn("=>{}不在线", clientId);
continue;
}
ResponseManager.publishClients(msg, session);
}
}
}
/**
* 更新客户端在线时间,给客户端发送消息时用这个看客户端最近是否在线
* 用来判断设备不正常掉线没有应答服务器的情况
*
* @param clientId 客户端id
*/
public static void updatePing(String clientId) {
pingMap.put(clientId, DateUtils.getTimestamp());
}
/**
* 平台判定设备状态 Ping客户端是否在线
*
* @param clientId 客户端id
* @return 结果
*/
public static Boolean validClient(String clientId) {
long currTime = DateUtils.getTimestamp();
/*获取客户端连接时,时间*/
Long timestamp = pingMap.get(clientId);
if (null == timestamp) {
return false;
}
//当设备缓存的心跳时间大于 平台判断时间 1.5f 表示设备不正常断开了服务器
if (currTime - timestamp > FastBeeConstant.SERVER.DEVICE_PING_EXPIRED) {
//pingMap.remove(clientId);
//SessionManger.removeClient(clientId);
return false;
}
return true;
}
}

View File

@@ -0,0 +1,77 @@
package com.fastbee.mqtt.manager;
import com.alibaba.fastjson2.JSON;
import com.fastbee.common.enums.DeviceStatus;
import com.fastbee.common.enums.TopicType;
import com.fastbee.common.utils.gateway.mq.TopicsUtils;
import com.fastbee.iot.domain.Device;
import com.fastbee.iot.service.IDeviceService;
import com.fastbee.mq.mqttClient.PubMqttClient;
import com.fastbee.mqtt.model.PushMessageBo;
import io.netty.buffer.Unpooled;
import io.netty.handler.codec.mqtt.*;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;
import javax.annotation.Resource;
import java.nio.charset.StandardCharsets;
@Component
public class MqttRemoteManager {
@Resource
private TopicsUtils topicsUtils;
@Resource
private IDeviceService deviceService;
/**
* true: 使用netty搭建的mqttBroker false: 使用emq
*/
@Value("${server.broker.enabled}")
private Boolean enabled;
@Resource
private PubMqttClient pubMqttClient;
/**
* 推送设备状态
* @param serialNumber 设备
* @param status 状态
*/
public void pushDeviceStatus(Long productId, String serialNumber, DeviceStatus status){
//兼容emqx推送TCP客户端上线
Device device = deviceService.selectDeviceNoModel(serialNumber);
String message = "{\"status\":" + status.getType() + ",\"isShadow\":" + device.getIsShadow() + ",\"rssi\":" + device.getRssi() + "}";
String topic = topicsUtils.buildTopic(device.getProductId(), serialNumber, TopicType.STATUS_POST);
if (enabled){
MqttPublishMessage publishMessage = (MqttPublishMessage) MqttMessageFactory.newMessage(
new MqttFixedHeader(MqttMessageType.PUBLISH, false, MqttQoS.AT_MOST_ONCE, false, 0),
new MqttPublishVariableHeader(topic, 0),
Unpooled.buffer().writeBytes(message.getBytes(StandardCharsets.UTF_8))
);
ClientManager.pubTopic(publishMessage);
}else {
//emqx直接用客户端推送
pubMqttClient.publish(1,false,topic,message);
}
}
/**
* 公共推送消息方法
* @param bo 消息体
*/
public void pushCommon(PushMessageBo bo){
//netty版本发送
if (enabled){
MqttPublishMessage publishMessage = (MqttPublishMessage) MqttMessageFactory.newMessage(
new MqttFixedHeader(MqttMessageType.PUBLISH, false, MqttQoS.AT_MOST_ONCE, false, 0),
new MqttPublishVariableHeader(bo.getTopic(), 0),
Unpooled.buffer().writeBytes(bo.getMessage().getBytes(StandardCharsets.UTF_8))
);
ClientManager.pubTopic(publishMessage);
}else {
pubMqttClient.publish(0,false,bo.getTopic(), bo.getMessage());
}
}
}

View File

@@ -0,0 +1,76 @@
package com.fastbee.mqtt.manager;
import com.fastbee.base.session.Session;
import io.netty.buffer.ByteBuf;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.handler.codec.mqtt.*;
import lombok.extern.slf4j.Slf4j;
@Slf4j
public class ResponseManager {
/**
* 发送信息:用于服务端收到消息客户端数据后,向客户端发送响应信息
*
* @param session 上下文
* @param msg mqtt消息
* @param flush 是否刷新
*/
public static void responseMessage(Session session, MqttMessage msg, boolean flush) {
ChannelFuture future = flush ? session.getHandlerContext().writeAndFlush(msg) : session.getHandlerContext().write(msg);
future.addListener(f -> {
if (!f.isSuccess()) {
log.error("=>响应设备[{}],发送消息:{},失败原因:{}", session.getClientId(), msg, f.cause());
}else {
//log.debug("=>相应设备:[{}],发送消息:[{}]",session.getClientId(),msg);
}
});
}
/**
* 发送信息:用于服务端向客户端通过clientID下发消息(单客户端)
*
* @param msg mqtt消息
* @param clientId 客户端id
* @param flush 是否刷新
*/
public static void sendMessage(MqttMessage msg, String clientId, boolean flush) {
Session session = SessionManger.getSession(clientId);
if (session == null || null == session.getHandlerContext()) {
return;
}
responseMessage(session, msg, flush);
}
/**
* 推送消息给订阅客户端(所有订阅客户端)
*
* @param msg 推送消息
* @param session 客户端
*/
public static void publishClients(MqttPublishMessage msg, Session session) {
try {
final Channel channel = session.getHandlerContext().channel();
MqttQoS qos = msg.fixedHeader().qosLevel();
ByteBuf sendBuf = msg.content().retainedDuplicate();
sendBuf.resetReaderIndex();
/*配置推送消息类型*/
MqttFixedHeader Header = new MqttFixedHeader(MqttMessageType.PUBLISH,
false, qos, msg.fixedHeader().isRetain(), 0);
/*设置topic packetId*/
MqttPublishVariableHeader publishVariableHeader = new MqttPublishVariableHeader(
msg.variableHeader().topicName(), msg.variableHeader().packetId());
/*推送消息*/
MqttPublishMessage publishMessage = new MqttPublishMessage(Header,
publishVariableHeader, sendBuf);
channel.writeAndFlush(publishMessage);
} catch (Exception e) {
log.error("=>发送消息异常 {}", msg, e);
}
}
}

View File

@@ -0,0 +1,53 @@
package com.fastbee.mqtt.manager;
import com.fastbee.mqtt.model.RetainMessage;
import io.netty.handler.codec.mqtt.MqttPublishMessage;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
@Slf4j
@Component
public class RetainMsgManager {
/*保存topic的retain消息*/
private static Map<String, RetainMessage> retainMap = new ConcurrentHashMap<>();
/**
* 推送保留信息到订阅客户端
*
* @param message 推送消息
*/
public static void pushMessage(MqttPublishMessage message) {
if (null == message || !message.fixedHeader().isRetain()) {
return;
}
byte[] bytes = new byte[message.payload().readableBytes()];
if (bytes.length > 0) {
RetainMessage retainMsg = RetainMessage.builder()
.topic(message.variableHeader().topicName())
.qos(message.fixedHeader().qosLevel().value()).message(bytes).build();
retainMap.put(message.variableHeader().topicName(), retainMsg);
} else {
retainMap.remove(message.variableHeader().topicName());
}
}
public static Integer getSize() {
return retainMap.size();
}
/**
* 获取消息
*
* @param topic 主题
* @return 消息
*/
public static RetainMessage getRetain(String topic) {
return retainMap.get(topic);
}
}

View File

@@ -0,0 +1,162 @@
package com.fastbee.mqtt.manager;
import com.fastbee.common.constant.FastBeeConstant;
import com.fastbee.common.core.mq.DeviceStatusBo;
import com.fastbee.common.enums.DeviceStatus;
import com.fastbee.common.exception.ServiceException;
import com.fastbee.common.utils.StringUtils;
import com.fastbee.common.utils.spring.SpringUtils;
import com.fastbee.iot.service.cache.IDeviceCache;
import com.fastbee.mq.redischannel.producer.MessageProducer;
import com.fastbee.mq.service.IMessagePublishService;
import com.fastbee.base.service.ISessionStore;
import com.fastbee.base.session.Session;
import com.fastbee.base.util.AttributeUtils;
import com.fastbee.mqtt.utils.MqttMessageUtils;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.mqtt.MqttMessageType;
import lombok.extern.slf4j.Slf4j;
import java.util.Map;
import java.util.Set;
/**
* 会话管理类
*
* @Author guanshubiao
* @Date 2022/9/12 20:22
*/
@Slf4j
public class SessionManger {
private static ISessionStore sessionStore = SpringUtils.getBean(ISessionStore.class);
private static MqttRemoteManager remoteManager = SpringUtils.getBean(MqttRemoteManager.class);
private static IDeviceCache deviceCache = SpringUtils.getBean(IDeviceCache.class);
/**
* mqtt新客户连接
*
* @param clientId 客户端id
* @param session 客户端
*/
public static void buildSession(String clientId, Session session) {
log.debug("=>新客户端连接clientId={}", clientId);
if (StringUtils.isEmpty(clientId) || handleContext(session)) {
log.error("=>客户端id为空或者session未注册!");
return;
}
sessionStore.storeSession(clientId, session);
//contextMap.put(session.getHandlerContext(), session);
/*更新客户端在平台的最新响应时间*/
ClientManager.updatePing(clientId);
/*发送MQ设备上线*/
DeviceStatusBo statusBo = MqttMessageUtils.buildStatusMsg(session.getHandlerContext(), session.getClientId(), DeviceStatus.ONLINE, session.getIp());
if (!statusBo.getSerialNumber().startsWith(FastBeeConstant.SERVER.WM_PREFIX) &&
!statusBo.getSerialNumber().startsWith(FastBeeConstant.SERVER.WS_PREFIX) &&
!statusBo.getSerialNumber().startsWith(FastBeeConstant.SERVER.FAST_PHONE)) {
deviceCache.updateDeviceStatusCache(statusBo);
remoteManager.pushDeviceStatus(-1L,statusBo.getSerialNumber(), statusBo.getStatus());
}
}
/**
* 根据客户端id移除客户端
*
* @param clientId 客户端id
*/
public static void removeClient(String clientId) {
log.debug("=>移除客户端,clientId={}", clientId);
try {
if (StringUtils.isEmpty(clientId) || !sessionStore.containsKey(clientId) || clientId.endsWith(FastBeeConstant.SERVER.WS_PREFIX) ||
clientId.endsWith(FastBeeConstant.SERVER.FAST_PHONE)) {
return;
}
Session session = sessionStore.getSession(clientId);
if (handleContext(session)) {
log.error("移除客户端失败,客户端未注册!");
return;
}
//关闭通道
session.getHandlerContext().close();
//移除client
sessionStore.cleanSession(clientId);
session.setMqttMessageType(MqttMessageType.DISCONNECT);
//发送至MQ,设备下线
DeviceStatusBo statusBo = MqttMessageUtils.buildStatusMsg(session.getHandlerContext(), session.getClientId(), DeviceStatus.OFFLINE, session.getIp());
if (!statusBo.getSerialNumber().startsWith(FastBeeConstant.SERVER.WM_PREFIX) &&
!statusBo.getSerialNumber().startsWith(FastBeeConstant.SERVER.WS_PREFIX)) {
deviceCache.updateDeviceStatusCache(statusBo);
remoteManager.pushDeviceStatus(-1L,statusBo.getSerialNumber(), statusBo.getStatus());
}
} catch (Exception e) {
throw new ServiceException("移除客户端失败,message=" + e.getMessage());
}
}
/**
* 根据客户通道移除客户端
*
* @param ctx 上下文通道
*/
public static void removeContextByContext(ChannelHandlerContext ctx) {
try {
/*获取*/
Session session = AttributeUtils.getSession(ctx.channel());
if (handleContext(session)) {
log.error("=>客户端通道不存在!移除失败");
return;
}
sessionStore.cleanSession(session.getClientId());
session.setMqttMessageType(MqttMessageType.DISCONNECT);
//发送至MQ,设备下线
DeviceStatusBo statusBo = MqttMessageUtils.buildStatusMsg(session.getHandlerContext(), session.getClientId(), DeviceStatus.OFFLINE, session.getIp());
if (!statusBo.getSerialNumber().startsWith(FastBeeConstant.SERVER.WM_PREFIX) &&
!statusBo.getSerialNumber().startsWith(FastBeeConstant.SERVER.WS_PREFIX)) {
deviceCache.updateDeviceStatusCache(statusBo);
remoteManager.pushDeviceStatus(-1L,statusBo.getSerialNumber(), statusBo.getStatus());
}
} catch (Exception e) {
log.error("=>移除客户端失败={}", e.getMessage());
}
}
/**
* ping判定时间超时
*
* @param clientId 客户id
*/
public static void pingTimeout(String clientId) {
try {
removeClient(clientId);
} catch (Exception e) {
throw new ServiceException("移除超时客户端失败");
}
}
/**
* 根据clientId获取客户通道
*
* @param clientId 客户端id
* @return session
*/
public static Session getSession(String clientId) {
return sessionStore.getSession(clientId);
}
/**
* 校验Session已经注册通道
*
* @param session 客户端
* @return 结果
*/
private static boolean handleContext(Session session) {
if (null == session || null == session.getHandlerContext()) {
return true;
}
return false;
}
}

View File

@@ -0,0 +1,30 @@
package com.fastbee.mqtt.manager;
import com.fastbee.mqtt.model.WillMessage;
import lombok.extern.slf4j.Slf4j;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
@Slf4j
public class WillMessageManager {
private static Map<String, WillMessage> map = new ConcurrentHashMap<>();
public static void push(WillMessage message){
map.put(message.getClientId(),message);
}
public static void pop(String clientId){
try {
WillMessage message = map.get(clientId);
if (null == message){
return;
}
ClientManager.pubTopic(message.getMessage());
}catch (Exception e){
log.error("=>发送客户端[{}],遗嘱消息异常",e.getMessage(),e);
}
}
}

View File

@@ -0,0 +1,45 @@
package com.fastbee.mqtt.model;
import io.netty.handler.codec.mqtt.MqttQoS;
import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
/**
* @author gsb
* @date 2022/10/7 19:04
*/
@Data
@NoArgsConstructor
@AllArgsConstructor
public class ClientMessage {
/*共享主题客户端id不为空则指定客户端发送*/
private String sharedClientId;
/*客户端id*/
private String clientId;
/*消息质量*/
private MqttQoS qos;
/*topic*/
private String topicName;
/*是否保留消息*/
private boolean retain;
/*数据*/
private byte[] payload;
private int messageId;
/*是否是遗嘱消息*/
private boolean willFlag;
/*是否是dup消息*/
private boolean dup;
public static ClientMessage of(MqttQoS qos,String topicName,boolean retain, byte[] payload){
return new ClientMessage(null,null,qos,topicName,retain,payload,0,false,false);
}
public static ClientMessage of(String clientId,MqttQoS qos,String topicName,boolean retain){
return new ClientMessage(null,clientId,qos,topicName,retain,null,0,false,false);
}
}

View File

@@ -0,0 +1,23 @@
package com.fastbee.mqtt.model;
import lombok.Data;
import java.io.Serializable;
/**
* @author bill
*/
@Data
public class PushMessageBo implements Serializable {
/*主题*/
private String topic;
/*数据*/
private String message;
/*消息质量*/
private int qos;
private Integer value;
private Integer address;
private Integer slaveId;
}

View File

@@ -0,0 +1,23 @@
package com.fastbee.mqtt.model;
import lombok.Builder;
import lombok.Data;
import java.io.Serializable;
/**
* 保留消息bo
* @author gsb
* @date 2022/9/16 14:05
*/
@Data
@Builder
public class RetainMessage implements Serializable {
/*主题*/
private String topic;
/*数据*/
private byte[] message;
/*消息质量*/
private int qos;
}

View File

@@ -0,0 +1,23 @@
package com.fastbee.mqtt.model;
import lombok.AllArgsConstructor;
import lombok.Data;
/**
* 订阅topic信息
* @author gsb
* @date 2022/10/14 8:30
*/
@Data
@AllArgsConstructor
public class Subscribe {
/*topic*/
private String topicName;
/*消息质量*/
private int qos;
/*客户端id*/
private String clientId;
/*清楚回话*/
private boolean cleanSession;
}

View File

@@ -0,0 +1,26 @@
package com.fastbee.mqtt.model;
import io.netty.handler.codec.mqtt.MqttPublishMessage;
import lombok.AllArgsConstructor;
import lombok.Data;
import java.io.Serializable;
/**
* @author gsb
* @date 2022/9/15 15:36
*/
@Data
@AllArgsConstructor
public class WillMessage implements Serializable {
private static final long serialVersionUID = -1L;
/*客户端Id*/
private String clientId;
/*清楚客户端*/
private boolean cleanSession;
/*topic*/
private String topic;
/*客户端推送消息*/
private MqttPublishMessage message;
}

View File

@@ -0,0 +1,66 @@
package com.fastbee.mqtt.server;
import com.fastbee.server.Server;
import com.fastbee.common.constant.FastBeeConstant;
import com.fastbee.mqtt.handler.adapter.MqttMessageAdapter;
import com.fastbee.server.config.NettyConfig;
import io.netty.bootstrap.AbstractBootstrap;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.mqtt.MqttDecoder;
import io.netty.handler.codec.mqtt.MqttEncoder;
import io.netty.handler.logging.LogLevel;
import io.netty.handler.logging.LoggingHandler;
import io.netty.handler.timeout.IdleStateHandler;
import io.netty.util.concurrent.DefaultThreadFactory;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
@Component
@Slf4j
public class MqttServer extends Server {
@Autowired
private MqttMessageAdapter messageAdapter;
@Override
protected AbstractBootstrap initialize() {
bossGroup = new NioEventLoopGroup(1, new DefaultThreadFactory(config.name, Thread.MAX_PRIORITY));
workerGroup = new NioEventLoopGroup(config.workerCore, new DefaultThreadFactory(config.name, Thread.MAX_PRIORITY));
if (config.businessCore > 0) {
businessService = new ThreadPoolExecutor(config.businessCore, config.businessCore, 1L,
TimeUnit.SECONDS, new LinkedBlockingQueue<>(), new DefaultThreadFactory(config.name, true, Thread.NORM_PRIORITY));
}
return new ServerBootstrap()
.group(bossGroup, workerGroup)
.channel(NioServerSocketChannel.class)
.handler(new LoggingHandler(LogLevel.DEBUG))
.option(ChannelOption.SO_BACKLOG, 511)
.childOption(ChannelOption.SO_KEEPALIVE, Boolean.TRUE)
.childHandler(new ChannelInitializer<NioSocketChannel>() {
@Override
protected void initChannel(NioSocketChannel channel) {
//客户端心跳检测机制
channel.pipeline()
.addFirst(FastBeeConstant.SERVER.IDLE
, new IdleStateHandler(config.readerIdleTime, config.writerIdleTime, config.allIdleTime, TimeUnit.SECONDS))
.addLast(FastBeeConstant.SERVER.DECODER, new MqttDecoder(1024 * 1024 * 2))
.addLast(FastBeeConstant.SERVER.ENCODER, MqttEncoder.INSTANCE)
.addLast(messageAdapter);
}
});
}
}

View File

@@ -0,0 +1,73 @@
package com.fastbee.mqtt.server;
import com.fastbee.server.Server;
import com.fastbee.common.constant.FastBeeConstant;
import com.fastbee.mqtt.codec.WebSocketMqttCodec;
import com.fastbee.mqtt.handler.adapter.MqttMessageAdapter;
import io.netty.bootstrap.AbstractBootstrap;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.codec.http.HttpContentCompressor;
import io.netty.handler.codec.http.HttpObjectAggregator;
import io.netty.handler.codec.http.HttpServerCodec;
import io.netty.handler.codec.http.websocketx.WebSocketServerProtocolHandler;
import io.netty.handler.codec.mqtt.MqttDecoder;
import io.netty.handler.codec.mqtt.MqttEncoder;
import io.netty.handler.timeout.IdleStateHandler;
import io.netty.util.concurrent.DefaultThreadFactory;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
/**
* @author gsb
* @date 2022/9/15 14:23
*/
@Component
@Slf4j
public class WebSocketServer extends Server {
@Autowired
private WebSocketMqttCodec webSocketMqttCodec;
@Autowired
private MqttMessageAdapter mqttMessageAdapter;
@Override
protected AbstractBootstrap initialize() {
bossGroup = new NioEventLoopGroup();
workerGroup = new NioEventLoopGroup();
return new ServerBootstrap()
.group(bossGroup, workerGroup)
.channel(NioServerSocketChannel.class)
.childHandler(new ChannelInitializer<SocketChannel>() {
@Override
public void initChannel(SocketChannel ch) {
ch.pipeline()
.addFirst(FastBeeConstant.WS.HEART_BEAT
, new IdleStateHandler(0, 0, 70))
/*http请求响应*/
.addLast(FastBeeConstant.WS.HTTP_SERVER_CODEC, new HttpServerCodec())
/*聚合header与body组成完整的Http请求最大数据量为1Mb*/
.addLast(FastBeeConstant.WS.AGGREGATOR, new HttpObjectAggregator(1024 * 1024))
/*压缩出站数据*/
.addLast(FastBeeConstant.WS.COMPRESSOR, new HttpContentCompressor())
/*WebSocket协议配置mqtt*/
.addLast(FastBeeConstant.WS.PROTOCOL, new WebSocketServerProtocolHandler("/mqtt",
"mqtt,mqttv3.1,mqttv3.1.1,mqttv5.0", true, 65536))
.addLast(FastBeeConstant.WS.MQTT_WEBSOCKET, webSocketMqttCodec)
.addLast(FastBeeConstant.WS.DECODER, new MqttDecoder())
.addLast(FastBeeConstant.WS.ENCODER, MqttEncoder.INSTANCE)
.addLast(FastBeeConstant.WS.BROKER_HANDLER, mqttMessageAdapter);
}
});
}
}

View File

@@ -0,0 +1,90 @@
package com.fastbee.mqtt.service;
import com.fastbee.mqtt.model.ClientMessage;
import java.util.List;
import java.util.Map;
/**
* @author gsb
* @date 2022/10/14 14:35
*/
public interface IMessageStore {
/**
* 存储控制包
*
* @param topic: 控制包所属主题
* @param clientMessage: 需要存储的消息
*/
void storeMessage(String topic, ClientMessage clientMessage);
/**
* 清除topic下的所有消息
*
* @param topic: 主题
*/
void cleanTopic(String topic);
/**
* 根据clientId清除消息
*
* @param clientId: 客户端唯一标识
*/
void removeMessage(String clientId);
/**
* 匹配主题过滤器,寻找对应消息
*
* @param topicFilter: 主题过滤器
*/
List<ClientMessage> searchMessages(String topicFilter);
/**
* 保存 clientMessage
*
* @param messageId 消息id
*/
public void savePubMsg(Integer messageId, ClientMessage clientMessage);
/**
* 移除
*
* @param messageId 消息id
*/
public void removePubMsg(int messageId);
/**
* 保存 REL IN
*
* @param messageId 消息id
*/
public void saveRelInMsg(int messageId);
/**
* 保存 REL OUT
*
* @param messageId 消息id
*/
public void saveRelOutMsg(int messageId);
/**
* 移除
*
* @param messageId 消息id
*/
public void removeRelInMsg(int messageId);
/**
* 移除
*
* @param messageId 消息id
*/
public void removeRelOutMsg(int messageId);
/**
* 判断Rel out是否包含消息id
*/
public boolean outRelContains(int messageId);
}

View File

@@ -0,0 +1,40 @@
package com.fastbee.mqtt.service;
import com.fastbee.mqtt.model.Subscribe;
import java.util.List;
/**
* 订阅缓存
* @author gsb
* @date 2022/10/14 8:24
*/
public interface ISubscriptionService {
/**
* 保存客户订阅的主题
*
* @param subscribeList 客户订阅
*/
void subscribe(List<Subscribe> subscribeList, String clientId);
/**
* 解除订阅
*
* @param clientId 客户id
* @param topicName 主题
*/
void unsubscribe(String clientId, String topicName);
/**
* 获取订阅了 topic 的客户id
*
* @param topic 主题
* @return 订阅了主题的客户id列表
*/
List<Subscribe> searchSubscribeClientList(String topic);
}

View File

@@ -0,0 +1,135 @@
package com.fastbee.mqtt.service.impl;
import com.alibaba.fastjson2.JSON;
import com.fastbee.common.enums.TopicType;
import com.fastbee.common.exception.ServiceException;
import com.fastbee.common.utils.DateUtils;
import com.fastbee.common.utils.gateway.mq.TopicsUtils;
import com.fastbee.iot.domain.Device;
import com.fastbee.iot.domain.EventLog;
import com.fastbee.common.core.thingsModel.ThingsModelSimpleItem;
import com.fastbee.common.core.thingsModel.ThingsModelValuesInput;
import com.fastbee.iot.service.IDeviceService;
import com.fastbee.iot.service.IEventLogService;
import com.fastbee.mq.model.ReportDataBo;
import com.fastbee.mq.mqttClient.PubMqttClient;
import com.fastbee.mq.service.IDataHandler;
import com.fastbee.mq.service.IMqttMessagePublish;
import com.fastbee.mqtt.manager.MqttRemoteManager;
import com.fastbee.mqtt.model.PushMessageBo;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;
import org.springframework.util.CollectionUtils;
import javax.annotation.Resource;
import java.util.ArrayList;
import java.util.List;
/**
* 上报数据处理方法集合
* @author bill
*/
@Service
@Slf4j
public class DataHandlerImpl implements IDataHandler {
@Resource
private IDeviceService deviceService;
@Resource
private IEventLogService eventLogService;
@Resource
private IMqttMessagePublish messagePublish;
@Resource
private MqttRemoteManager remoteManager;
@Resource
private TopicsUtils topicsUtils;
/**
* 上报属性或功能处理
*
* @param bo 上报数据模型
*/
@Override
public void reportData(ReportDataBo bo) {
try {
List<ThingsModelSimpleItem> thingsModelSimpleItems = bo.getDataList();
if (CollectionUtils.isEmpty(bo.getDataList()) || bo.getDataList().size() == 0) {
thingsModelSimpleItems = JSON.parseArray(bo.getMessage(), ThingsModelSimpleItem.class);
}
ThingsModelValuesInput input = new ThingsModelValuesInput();
input.setProductId(bo.getProductId());
input.setDeviceNumber(bo.getSerialNumber().toUpperCase());
input.setThingsModelValueRemarkItem(thingsModelSimpleItems);
input.setSlaveId(bo.getSlaveId());
List<ThingsModelSimpleItem> result = deviceService.reportDeviceThingsModelValue(input, bo.getType(), bo.isShadow());
} catch (Exception e) {
log.error("接收属性数据,解析数据时异常 message={},e={}", e.getMessage(),e);
}
}
/**
* 上报事件
*
* @param bo 上报数据模型
*/
@Override
public void reportEvent(ReportDataBo bo) {
try {
List<ThingsModelSimpleItem> thingsModelSimpleItems = JSON.parseArray(bo.getMessage(), ThingsModelSimpleItem.class);
Device device = deviceService.selectDeviceBySerialNumber(bo.getSerialNumber());
List<EventLog> results = new ArrayList<>();
for (int i = 0; i < thingsModelSimpleItems.size(); i++) {
// 添加到设备日志
EventLog event = new EventLog();
event.setDeviceId(device.getDeviceId());
event.setDeviceName(device.getDeviceName());
event.setLogValue(thingsModelSimpleItems.get(i).getValue());
event.setRemark(thingsModelSimpleItems.get(i).getRemark());
event.setSerialNumber(device.getSerialNumber());
event.setIdentity(thingsModelSimpleItems.get(i).getId());
event.setLogType(3);
event.setIsMonitor(0);
event.setUserId(device.getUserId());
event.setUserName(device.getUserName());
event.setTenantId(device.getTenantId());
event.setTenantName(device.getTenantName());
event.setCreateTime(DateUtils.getNowDate());
// 1=影子模式2=在线模式3=其他
event.setMode(2);
results.add(event);
//eventLogService.insertEventLog(event);
}
eventLogService.insertBatch(results);
} catch (Exception e) {
log.error("接收事件,解析数据时异常 message={}", e.getMessage());
}
}
/**
* 上报设备信息
*/
public void reportDevice(ReportDataBo bo) {
try {
// 设备实体
Device deviceEntity = deviceService.selectDeviceBySerialNumber(bo.getSerialNumber());
// 上报设备信息
Device device = JSON.parseObject(bo.getMessage(), Device.class);
device.setProductId(bo.getProductId());
device.setSerialNumber(bo.getSerialNumber());
deviceService.reportDevice(device, deviceEntity);
// 发布设备状态
messagePublish.publishStatus(bo.getProductId(), bo.getSerialNumber(), 3, deviceEntity.getIsShadow(), device.getRssi());
} catch (Exception e) {
log.error("接收设备信息,解析数据时异常 message={}", e.getMessage());
throw new ServiceException(e.getMessage(), 1);
}
}
}

View File

@@ -0,0 +1,141 @@
package com.fastbee.mqtt.service.impl;
import com.fastbee.common.core.mq.DeviceReport;
import com.fastbee.common.core.mq.DeviceReportBo;
import com.fastbee.common.core.mq.message.DeviceData;
import com.fastbee.common.enums.ServerType;
import com.fastbee.common.enums.ThingsModelType;
import com.fastbee.common.enums.TopicType;
import com.fastbee.common.exception.ServiceException;
import com.fastbee.common.utils.gateway.mq.TopicsUtils;
import com.fastbee.iot.domain.Device;
import com.fastbee.iot.service.IDeviceService;
import com.fastbee.json.JsonProtocolService;
import com.fastbee.mq.model.ReportDataBo;
import com.fastbee.mq.service.IDataHandler;
import com.fastbee.mq.service.IDeviceReportMessageService;
import io.netty.buffer.Unpooled;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import javax.annotation.Resource;
import java.util.Optional;
/**
* 处理类 处理设备主动上报和设备回调信息
*
* @author bill
*/
@Service
@Slf4j
public class DeviceReportMessageServiceImpl implements IDeviceReportMessageService {
@Autowired
private IDeviceService deviceService;
@Autowired
private JsonProtocolService jsonProtocolService;
@Resource
private TopicsUtils topicsUtils;
@Resource
private IDataHandler dataHandler;
/**
* 处理设备主动上报数据
*/
@Override
public void parseReportMsg(DeviceReportBo bo) {
if (bo.getServerType() == ServerType.MQTT) {
//构建消息
Device report = buildReport(bo);
/*获取协议处理器*/
DeviceData data = DeviceData.builder()
.serialNumber(bo.getSerialNumber())
.topicName(bo.getTopicName())
.productId(report.getProductId())
.data(bo.getData())
.prop(bo.getProp())
.buf(Unpooled.wrappedBuffer(bo.getData()))
.build();
/*根据协议解析后的数据*/
DeviceReport reportMessage = jsonProtocolService.decode(data, null);
reportMessage.setSerialNumber(bo.getSerialNumber());
reportMessage.setProductId(bo.getProductId());
reportMessage.setPlatformDate(bo.getPlatformDate());
reportMessage.setServerType(bo.getServerType());
reportMessage.setUserId(report.getUserId());
reportMessage.setUserName(report.getUserName());
reportMessage.setDeviceName(report.getDeviceName());
processNoSub(reportMessage, bo.getTopicName());
}
}
/**
* 构建消息
*
* @param bo
*/
@Override
public Device buildReport(DeviceReportBo bo) {
String serialNumber = topicsUtils.parseSerialNumber(bo.getTopicName());
Device device = deviceService.selectDeviceBySerialNumber(serialNumber);
Optional.ofNullable(device).orElseThrow(() -> new ServiceException("设备不存在"));
//设置物模型
String thingsModel = topicsUtils.getThingsModel(bo.getTopicName());
ThingsModelType thingsModelType = ThingsModelType.getType(thingsModel);
bo.setType(thingsModelType);
//产品id
bo.setProductId(device.getProductId());
//设备编号
bo.setSerialNumber(serialNumber);
return device;
}
/**
* 处理网关设备
*
* @param message
* @param topicName
*/
private void processNoSub(DeviceReport message, String topicName) {
//处理设备上报数据
handlerReportMessage(message, topicName);
}
/**
* 处理设备主动上报属性
*
* @param topicName
* @param message
*/
public void handlerReportMessage(DeviceReport message, String topicName) {
if (message.getServerType().equals(ServerType.MQTT)){
//处理topic以prop结尾上报的数据 (属性)
if (message.getServerType().equals(ServerType.MQTT)) {
if (!topicName.endsWith(TopicType.PROPERTY_POST.getTopicSuffix())
&& !topicName.endsWith(TopicType.PROPERTY_POST_SIMULATE.getTopicSuffix())) {
return;
}
}
}
ReportDataBo report = new ReportDataBo();
report.setSerialNumber(message.getSerialNumber());
report.setProductId(message.getProductId());
report.setDataList(message.getValuesInput().getThingsModelValueRemarkItem());
report.setType(1);
report.setSlaveId(message.getSlaveId());
report.setUserId(message.getUserId());
report.setUserName(message.getUserName());
report.setDeviceName(message.getDeviceName());
//属性上报执行规则引擎
report.setRuleEngine(true);
dataHandler.reportData(report);
}
}

View File

@@ -0,0 +1,159 @@
package com.fastbee.mqtt.service.impl;
import com.fastbee.common.utils.gateway.mq.TopicsUtils;
import com.fastbee.mqtt.model.ClientMessage;
import com.fastbee.mqtt.service.IMessageStore;
import org.springframework.stereotype.Service;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
/**
* Retain will Qos12消息存储接口 -TODO 后续Redis处理
*
* @author gsb
* @date 2022/10/14 14:35
*/
@Service
public class MessageStoreImpl implements IMessageStore {
/**
* 存储消息,保留消息,遗留消息
*/
private final Map<String, ClientMessage> willOrRetainMap = new ConcurrentHashMap<>();
/**
* Qos2 Pub消息
*/
private final Map<Integer, ClientMessage> publishMap = new ConcurrentHashMap<>();
/**
* Qos2 REL IN消息
*/
private final Set<Integer> outRelSet = new HashSet<>();
/**
* Qos2 REL out
*/
private final Set<Integer> inRelSet = new HashSet<>();
/**
* 存储控制包
*
* @param topic: 控制包所属主题
* @param clientMessage: 需要存储的消息
*/
@Override
public void storeMessage(String topic, ClientMessage clientMessage) {
willOrRetainMap.put(topic, clientMessage);
}
/**
* 清除topic下的所有消息
*
* @param topic: 主题
*/
@Override
public void cleanTopic(String topic) {
willOrRetainMap.remove(topic);
}
/**
* 根据clientId清除消息
*
* @param clientId: 客户端唯一标识
*/
@Override
public void removeMessage(String clientId) {
for (Map.Entry<String, ClientMessage> entry : willOrRetainMap.entrySet()) {
if (entry.getValue().getClientId().equals(clientId)) {
willOrRetainMap.remove(entry.getKey());
}
}
}
/**
* 匹配主题过滤器,匹配消息
*
* @param topicFilter: 主题过滤器
*/
@Override
public List<ClientMessage> searchMessages(String topicFilter) {
List<ClientMessage> messageList = new ArrayList<>();
for (String topic : willOrRetainMap.keySet()) {
if (TopicsUtils.matchTopic(topic, topicFilter)) {
messageList.add(willOrRetainMap.get(topic));
}
}
return messageList;
}
/**
* 保存 clientMessage
*
* @param messageId 消息id
*/
@Override
public void savePubMsg(Integer messageId, ClientMessage clientMessage){
publishMap.put(messageId,clientMessage);
}
/**
* 移除
*
* @param messageId 消息id
*/
@Override
public void removePubMsg(int messageId){
publishMap.remove(messageId);
}
/**
* 保存 REL IN
*
* @param messageId 消息id
*/
@Override
public void saveRelInMsg(int messageId){
inRelSet.add(messageId);
}
/**
* 保存 REL OUT
*
* @param messageId 消息id
*/
@Override
public void saveRelOutMsg(int messageId){
outRelSet.add(messageId);
}
/**
* 移除
*
* @param messageId 消息id
*/
@Override
public void removeRelInMsg(int messageId){
inRelSet.remove(messageId);
}
/**
* 移除
*
* @param messageId 消息id
*/
@Override
public void removeRelOutMsg(int messageId){
outRelSet.remove(messageId);
}
/**
* 判断Rel out是否包含消息id
*/
@Override
public boolean outRelContains(int messageId){
return outRelSet.contains(messageId);
}
}

View File

@@ -0,0 +1,354 @@
package com.fastbee.mqtt.service.impl;
import com.alibaba.fastjson2.JSON;
import com.alibaba.fastjson2.JSONArray;
import com.fastbee.common.core.mq.DeviceReportBo;
import com.fastbee.common.core.mq.MQSendMessageBo;
import com.fastbee.common.core.mq.message.DeviceData;
import com.fastbee.common.core.mq.message.DeviceDownMessage;
import com.fastbee.common.core.mq.message.InstructionsMessage;
import com.fastbee.common.core.mq.message.MqttBo;
import com.fastbee.common.core.mq.ota.OtaUpgradeBo;
import com.fastbee.common.core.protocol.modbus.ModbusCode;
import com.fastbee.common.core.thingsModel.ThingsModelSimpleItem;
import com.fastbee.common.enums.ServerType;
import com.fastbee.common.enums.TopicType;
import com.fastbee.common.exception.ServiceException;
import com.fastbee.common.utils.DateUtils;
import com.fastbee.common.utils.StringUtils;
import com.fastbee.common.utils.gateway.CRC16Utils;
import com.fastbee.common.utils.gateway.mq.TopicsUtils;
import com.fastbee.common.utils.ip.IpUtils;
import com.fastbee.iot.domain.Device;
import com.fastbee.iot.domain.FunctionLog;
import com.fastbee.iot.domain.Product;
import com.fastbee.iot.model.NtpModel;
import com.fastbee.iot.model.ThingsModels.PropertyDto;
import com.fastbee.iot.service.IDeviceService;
import com.fastbee.iot.service.IProductService;
import com.fastbee.iot.service.IThingsModelService;
import com.fastbee.iot.service.cache.IFirmwareCache;
import com.fastbee.iot.util.SnowflakeIdWorker;
import com.fastbee.json.JsonProtocolService;
import com.fastbee.mq.model.ReportDataBo;
import com.fastbee.mq.mqttClient.PubMqttClient;
import com.fastbee.mq.service.IDataHandler;
import com.fastbee.mq.service.IMqttMessagePublish;
import com.fastbee.mqtt.manager.MqttRemoteManager;
import com.fastbee.mqtt.model.PushMessageBo;
import lombok.extern.slf4j.Slf4j;
import org.java_websocket.protocols.IProtocol;
import org.springframework.stereotype.Service;
import javax.annotation.Resource;
import java.util.ArrayList;
import java.util.List;
import java.util.Objects;
import java.util.Optional;
/**
* 消息推送方法集合
*
* @author bill
*/
@Slf4j
@Service
public class MqttMessagePublishImpl implements IMqttMessagePublish {
@Resource
private IProductService productService;
@Resource
private PubMqttClient mqttClient;
@Resource
private IFirmwareCache firmwareCache;
@Resource
private TopicsUtils topicsUtils;
@Resource
private IDeviceService deviceService;
@Resource
private MqttRemoteManager remoteManager;
@Resource
private IDataHandler dataHandler;
@Resource
private IThingsModelService thingsModelService;
@Resource
private JsonProtocolService jsonProtocolService;
private SnowflakeIdWorker snowflakeIdWorker = new SnowflakeIdWorker(3);
@Override
public InstructionsMessage buildMessage(DeviceDownMessage downMessage, TopicType type) {
/*返回的组将数据*/
InstructionsMessage message = new InstructionsMessage();
/*根据设备编号查询产品信息*/
if (StringUtils.isEmpty(downMessage.getProtocolCode())) {
Product product = productService.getProductBySerialNumber(downMessage.getSerialNumber());
Optional.ofNullable(product).orElseThrow(() -> new ServiceException("产品为空"));
downMessage.setProtocolCode(product.getProtocolCode());
}
String serialNumber = downMessage.getSerialNumber() == null ? "" : downMessage.getSerialNumber();
/*组建Topic*/
String topicName = "";
if (downMessage.getServerType().equals(ServerType.MQTT)) {
topicName = topicsUtils.buildTopic(downMessage.getProductId(), serialNumber, type);
}
DeviceData encodeData = DeviceData.builder()
.downMessage(downMessage)
.serialNumber(serialNumber)
.body(downMessage.getBody())
.code(downMessage.getCode())
.topicName(topicName).build();
//根据协议编码后数据
byte[] data = jsonProtocolService.encode(encodeData, null);
message.setMessage(data);
message.setSerialNumber(serialNumber);
message.setTopicName(topicName);
return message;
}
/**
* 服务(指令)下发
*/
@Override
public void funcSend(MQSendMessageBo bo) {
//如果协议编号为空,则获取
if (StringUtils.isEmpty(bo.getProtocolCode())) {
Product product = productService.selectProductByProductId(bo.getProductId());
//bo.setType(ThingsModelType.SERVICE);
bo.setProtocolCode(product.getProtocolCode());
bo.setTransport(product.getTransport());
}
//处理设备影子模式
if (null != bo.getIsShadow() && bo.getIsShadow()){
List<ThingsModelSimpleItem> dataList = new ArrayList<>();
bo.getValue().forEach((key,value) ->{
ThingsModelSimpleItem item = new ThingsModelSimpleItem();
item.setId(key);
item.setValue(value+"");
dataList.add(item);
});
ReportDataBo dataBo = new ReportDataBo();
dataBo.setDataList(dataList);
dataBo.setProductId(bo.getProductId());
dataBo.setSerialNumber(bo.getSerialNumber());
dataBo.setRuleEngine(false);
dataBo.setShadow(true);
dataBo.setSlaveId(bo.getSlaveId());
dataBo.setType(bo.getType().getCode());
dataHandler.reportData(dataBo);
return;
}
/* 下发服务数据存储对象*/
FunctionLog log = new FunctionLog();
log.setCreateTime(DateUtils.getNowDate());
log.setFunValue(bo.getValue().get(bo.getIdentifier()).toString());
log.setMessageId(bo.getMessageId());
log.setSerialNumber(bo.getSerialNumber());
log.setIdentify(bo.getIdentifier());
log.setShowValue(bo.getShowValue());
log.setFunType(1);
log.setModelName(bo.getModelName());
//兼容子设备
if (null != bo.getSlaveId()) {
PropertyDto thingModels = thingsModelService.getSingleThingModels(bo.getProductId(), bo.getIdentifier() + "#" + bo.getSlaveId());
log.setSerialNumber(bo.getSerialNumber() + "_" + bo.getSlaveId());
bo.setCode(ModbusCode.Write06);
if (!Objects.isNull(thingModels.getCode())){
bo.setCode(ModbusCode.getInstance(Integer.parseInt(thingModels.getCode())));
}
}
ServerType serverType = ServerType.explain(bo.getTransport());
Optional.ofNullable(serverType).orElseThrow(() -> new ServiceException("产品的传输协议编码为空!"));
/*下发服务数据处理对象*/
DeviceDownMessage downMessage = DeviceDownMessage.builder()
.messageId(bo.getMessageId())
.body(bo.getValue())
.serialNumber(bo.getSerialNumber())
.productId(bo.getProductId())
.timestamp(DateUtils.getTimestamp())
.identifier(bo.getIdentifier())
.slaveId(bo.getSlaveId())
.code(bo.getCode() == ModbusCode.Read01 ? ModbusCode.Write05 : ModbusCode.Write06)
.serverType(serverType)
.build();
switch (serverType) {
case MQTT:
//组建下发服务指令
InstructionsMessage instruction = buildMessage(downMessage, TopicType.FUNCTION_GET);
mqttClient.publish(instruction.getTopicName(), instruction.getMessage(), log);
MqttMessagePublishImpl.log.debug("=>服务下发,topic=[{}],指令=[{}]", instruction.getTopicName(),new String(instruction.getMessage()));
break;
}
}
/**
* OTA升级下发
*
* @param bo
*/
@Override
public void upGradeOTA(OtaUpgradeBo bo) {
}
@Override
public void sendFunctionMessage(DeviceReportBo bo) {
log.warn("=>功能指令下发,sendFunctionMessage bo=[{}]", bo);
Device device = deviceService.selectDeviceBySerialNumber(bo.getSerialNumber());
Optional.ofNullable(device).orElseThrow(()->new ServiceException("服务下发的设备:["+bo.getSerialNumber()+"]不存在"));
Product product = productService.selectProductByProductId(topicsUtils.parseProductId(bo.getTopicName()));
ServerType serverType = ServerType.explain(product.getTransport());
Optional.ofNullable(serverType).orElseThrow(() -> new ServiceException("产品的传输协议编码为空!"));
switch (serverType) {
case GB28181:
break;
}
}
/**
* 1.发布设备状态
*/
@Override
public void publishStatus(Long productId, String deviceNum, int deviceStatus, int isShadow, int rssi) {
String message = "{\"status\":" + deviceStatus + ",\"isShadow\":" + isShadow + ",\"rssi\":" + rssi + "}";
String topic = topicsUtils.buildTopic(productId, deviceNum, TopicType.STATUS_POST);
mqttClient.publish(1, false, topic, message);
}
/**
* 2.发布设备信息
*/
@Override
public void publishInfo(Long productId, String deviceNum) {
String topic = topicsUtils.buildTopic(productId, deviceNum, TopicType.INFO_GET);
mqttClient.publish(1, false, topic, "");
}
/**
* 3.发布时钟同步信息
*
* @param bo 数据模型
*/
public void publishNtp(ReportDataBo bo) {
NtpModel ntpModel = JSON.parseObject(bo.getMessage(), NtpModel.class);
ntpModel.setServerRecvTime(System.currentTimeMillis());
ntpModel.setServerSendTime(System.currentTimeMillis());
String topic = topicsUtils.buildTopic(bo.getProductId(), bo.getSerialNumber(), TopicType.NTP_GET);
mqttClient.publish(1, false, topic, JSON.toJSONString(ntpModel));
}
/**
* 4.发布属性
* delay 延时,秒为单位
*/
@Override
public void publishProperty(Long productId, String deviceNum, List<ThingsModelSimpleItem> thingsList, int delay) {
String pre = "";
if (delay > 0) {
pre = "$delayed/" + String.valueOf(delay) + "/";
}
String topic = topicsUtils.buildTopic(productId, deviceNum, TopicType.FUNCTION_GET);
if (thingsList == null) {
mqttClient.publish(1, true, topic, "");
} else {
mqttClient.publish(1, true, topic, JSON.toJSONString(thingsList));
}
}
/**
* 5.发布功能
* delay 延时,秒为单位
*/
@Override
public void publishFunction(Long productId, String deviceNum, List<ThingsModelSimpleItem> thingsList, int delay) {
String pre = "";
if (delay > 0) {
pre = "$delayed/" + String.valueOf(delay) + "/";
}
String topic = topicsUtils.buildTopic(productId, deviceNum, TopicType.FUNCTION_GET);
if (thingsList == null) {
mqttClient.publish(1, true, topic, "");
} else {
mqttClient.publish(1, true, topic, JSON.toJSONString(thingsList));
}
}
/**
* 设备数据同步
*
* @param deviceNumber 设备编号
* @return 设备
*/
public Device deviceSynchronization(String deviceNumber) {
Device device = deviceService.selectDeviceBySerialNumber(deviceNumber);
// 1-未激活2-禁用3-在线4-离线
if (device.getStatus() == 3) {
device.setStatus(4);
deviceService.updateDeviceStatus(device);
// 发布设备信息
publishInfo(device.getProductId(), device.getSerialNumber());
}
return device;
}
/**
* 发送模拟设备到WS
*/
public void sendSimulationWs(MqttBo send ,MqttBo receive,String topic){
PushMessageBo messageBo = new PushMessageBo();
messageBo.setTopic(topic);
JSONArray array = new JSONArray();
send.setDirection("send");
send.setTs(DateUtils.getNowDate());
receive.setTs(DateUtils.getNowDate());
receive.setDirection("receive");
array.add(send);
array.add(receive);
messageBo.setMessage(array.toJSONString());
remoteManager.pushCommon(messageBo);
}
public byte[] CRC(byte[] source) {
source[2] = (byte)((int) source[2] * 2);
byte[] result = new byte[source.length + 2];
byte[] crc16Byte = CRC16Utils.getCrc16Byte(source);
System.arraycopy(source, 0, result, 0, source.length);
System.arraycopy(crc16Byte, 0, result, result.length - 2, 2);
return result;
}
/**
* 搭建消息
*
* @param bo
* @return
*/
private DeviceDownMessage buildMessage(OtaUpgradeBo bo) {
String messageId = String.valueOf(snowflakeIdWorker.nextId());
bo.setMessageId(messageId);
bo.setOtaUrl("http://" + IpUtils.getHostIp()+bo.getOtaUrl());
return DeviceDownMessage.builder()
.productId(bo.getProductId())
.serialNumber(bo.getSerialNumber())
.body(JSON.toJSON(bo))
.timestamp(DateUtils.getTimestamp())
.messageId(messageId)
.build();
}
}

View File

@@ -0,0 +1,56 @@
package com.fastbee.mqtt.service.impl;
import com.fastbee.common.core.redis.RedisCache;
import com.fastbee.mqtt.model.Subscribe;
import com.fastbee.mqtt.service.ISubscriptionService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import java.util.List;
/**
* @author gsb
* @date 2022/10/14 8:37
*/
@Slf4j
@Component
public class SubscriptionServiceImpl implements ISubscriptionService {
@Autowired
private RedisCache redisCache;
/**
* 保存客户订阅的主题
*
* @param subscribeList 主题列表
*/
@Override
public void subscribe(List<Subscribe> subscribeList, String clientId) {
redisCache.setCacheList(clientId, subscribeList);
}
/**
* 解除订阅
*
* @param clientId 客户id
* @param topicName 主题
*/
@Override
public void unsubscribe(String clientId, String topicName) {
redisCache.delHashValue(topicName, clientId);
}
/**
* 获取订阅了 topic 的客户id
*
* @param topic 主题
* @return 订阅了主题的客户id列表
*/
@Override
public List<Subscribe> searchSubscribeClientList(String topic) {
return null;
}
}

View File

@@ -0,0 +1,144 @@
package com.fastbee.mqtt.utils;
import com.fastbee.common.core.mq.DeviceStatusBo;
import com.fastbee.common.enums.DeviceStatus;
import com.fastbee.common.utils.DateUtils;
import com.fastbee.mqtt.model.ClientMessage;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.mqtt.*;
import java.net.InetSocketAddress;
import java.util.ArrayList;
import java.util.List;
import java.util.Set;
import java.util.stream.Collectors;
/**
* 服务器应答信息构建
* @author gsb
* @date 2022/10/7 14:17
*/
public class MqttMessageUtils {
/**
* 服务器确认连接应答消息 CONNACK
*/
public static MqttConnAckMessage buildConntAckMessage(MqttConnectReturnCode code, boolean sessionPresent) {
MqttFixedHeader fixedHeader = buildFixedHeader(MqttMessageType.CONNACK);
MqttConnAckVariableHeader variableHeader = new MqttConnAckVariableHeader(code, sessionPresent);
return new MqttConnAckMessage(fixedHeader, variableHeader);
}
/**
* 设备ping(心跳信息)应答 PINGRESP
*/
public static MqttMessage buildPingResp() {
MqttFixedHeader fixedHeader = buildFixedHeader(MqttMessageType.PINGRESP);
return new MqttMessage(fixedHeader);
}
/**
* 取消订阅消息应答 UNSUBACK
*/
public static MqttUnsubAckMessage buildUnsubAckMessage(MqttMessage message) {
/*构建固定报文*/
MqttFixedHeader fixedHeader = buildFixedHeader(MqttMessageType.UNSUBACK);
return new MqttUnsubAckMessage(fixedHeader, getIdVariableHeader(message));
}
/**
* 订阅确认应答 SUBACK
*/
public static MqttSubAckMessage buildSubAckMessage(MqttMessage message) {
/*构建固定报文*/
MqttFixedHeader fixedHeader = buildFixedHeader(MqttMessageType.SUBACK);
/*构建可变报文*/
MqttSubscribeMessage mqttSubscribeMessage = (MqttSubscribeMessage) message;
/*获取订阅topic的Qos*/
Set<String> topics = mqttSubscribeMessage.payload().topicSubscriptions().stream().map(MqttTopicSubscription::topicName).collect(Collectors.toSet());
List<Integer> grantedQos = new ArrayList<>(topics.size());
for (int i = 0; i < topics.size(); i++) {
grantedQos.add(mqttSubscribeMessage.payload().topicSubscriptions().get(i).qualityOfService().value());
}
/*负载*/
MqttSubAckPayload payload = new MqttSubAckPayload(grantedQos);
return new MqttSubAckMessage(fixedHeader, getIdVariableHeader(message), payload);
}
/**
* 构建推送应答消息 PUBLISH
*/
public static MqttPublishMessage buildPublishMessage(ClientMessage msg, int packageId) {
/*报文固定头*/
MqttFixedHeader fixedHeader = new MqttFixedHeader(MqttMessageType.PUBLISH, msg.isDup(), msg.getQos(), false, 0);
/*报文可变头*/
MqttPublishVariableHeader variableHeader = new MqttPublishVariableHeader(msg.getTopicName(), packageId);
/*负载*/
ByteBuf payload = msg.getPayload() == null ? Unpooled.EMPTY_BUFFER : Unpooled.wrappedBuffer(msg.getPayload());
/*完整报文,固定头+可变头+payload*/
return new MqttPublishMessage(fixedHeader, variableHeader, payload);
}
/**
* Qos1 收到发布消息确认 无负载 PUBACK
*/
public static MqttPubAckMessage buildAckMessage(MqttMessage message) {
MqttFixedHeader fixedHeader = buildFixedHeader(MqttMessageType.PUBACK);
return new MqttPubAckMessage(fixedHeader, getIdVariableHeader(message));
}
/**
* Qos2 发到消息收到 无负载 PUBREC
*/
public static MqttMessage buildPubRecMessage(MqttMessage message){
MqttFixedHeader fixedHeader = buildFixedHeader(MqttMessageType.PUBREC);
return new MqttMessage(fixedHeader, getIdVariableHeader(message));
}
/**
* Qos2 发布消息释放 PUBREL
*/
public static MqttMessage buildPubRelMessage(MqttMessage message){
MqttFixedHeader fixedHeader = buildFixedHeader(MqttMessageType.PUBREL);
return new MqttMessage(fixedHeader, getIdVariableHeader(message));
}
/**
* Qos2 发布消息完成 PUBCOMP
*/
public static MqttMessage buildPubCompMessage(MqttMessage message){
MqttFixedHeader fixedHeader = buildFixedHeader(MqttMessageType.PUBCOMP);
return new MqttMessage(fixedHeader, getIdVariableHeader(message));
}
/**
* 固定头定制
*/
public static MqttFixedHeader buildFixedHeader(MqttMessageType messageType) {
return new MqttFixedHeader(messageType, false, MqttQoS.AT_MOST_ONCE, false, 0);
}
/**
* 构造MqttMessageIdVariableHeader
*/
public static MqttMessageIdVariableHeader getIdVariableHeader(MqttMessage mqttMessage) {
MqttMessageIdVariableHeader idVariableHeader = (MqttMessageIdVariableHeader) mqttMessage.variableHeader();
return MqttMessageIdVariableHeader.from(idVariableHeader.messageId());
}
/*构造返回MQ的设备状态model*/
public static DeviceStatusBo buildStatusMsg(ChannelHandlerContext ctx, String clientId,DeviceStatus status,String ip){
return DeviceStatusBo.builder()
.serialNumber(clientId)
.status(status)
.ip(ip)
.hostName(ip)
.timestamp(DateUtils.getNowDate()).build();
}
}

View File

@@ -0,0 +1,54 @@
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<parent>
<artifactId>fastbee</artifactId>
<groupId>com.fastbee</groupId>
<version>3.8.5</version>
</parent>
<packaging>pom</packaging>
<artifactId>fastbee-server</artifactId>
<description>服务集成模块</description>
<modules>
<!--服务基础模块-->
<module>base-server</module>
<!--服务启动模块-->
<module>boot-strap</module>
<!--mqttBroker-->
<module>mqtt-broker</module>
<!--服务核心tcp udp服务搭建模块-->
<module>iot-server-core</module>
</modules>
<dependencies>
<dependency>
<groupId>io.netty</groupId>
<artifactId>netty-all</artifactId> <!-- Use 'netty-all' for 4.0 or above -->
<version>4.1.56.Final</version>
<scope>compile</scope>
</dependency>
<dependency>
<groupId>com.fastbee</groupId>
<artifactId>fastbee-common</artifactId>
<version>3.8.5</version>
</dependency>
<dependency>
<groupId>org.projectlombok</groupId>
<artifactId>lombok</artifactId>
</dependency>
<dependency>
<groupId>com.fastbee</groupId>
<artifactId>fastbee-iot-service</artifactId>
</dependency>
</dependencies>
</project>