autumn/jsowell-charger-web
1
0
Fork 0
mirror of https://codeup.aliyun.com/67c68d4e484ca2f0a13ac3c1/ydc/jsowell-charger-web.git synced 2026-04-20 11:05:18 +08:00
Code Issues Packages Projects Releases Wiki Activity

Merge branch 'dev-new' into dev-new-rabbitmq

Browse Source 
# Conflicts:
#	jsowell-netty/src/main/java/com/jsowell/netty/server/yunkuaichong/NettyServerChannelInitializer.java
This commit is contained in:
Guoqs
2024-11-27 16:58:36 +08:00
parent 8493d49438 e861eec868
commit b345d8392c
12 changed files with 563 additions and 16 deletions
Download Patch File Download Diff File Expand all files Collapse all files

105
jsowell-netty/src/main/java/com/jsowell/netty/decoder/YkcProtocolDecoder.java Normal file
View File

@@ -0,0 +1,105 @@
package com.jsowell.netty.decoder;
import com.jsowell.common.util.BytesUtil;
import com.jsowell.common.util.CRC16Util;
import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.ByteToMessageDecoder;
import io.netty.handler.codec.DecoderException;
import lombok.extern.slf4j.Slf4j;
import java.util.List;
@Slf4j
public class YkcProtocolDecoder extends ByteToMessageDecoder {
@Override
protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception {
log.info("YkcProtocolDecoder.decode");
// 检查起始标志是否为0x68
if (in.readableBytes() < 5) {
return; // 至少需要 起始标志 (1) + 数据长度 (1) + 序列号域 (2) + 帧校验域 (1)
}
in.markReaderIndex(); // 标记当前读取位置
byte startFlag = in.readByte();
log.info("startFlag: {}", BytesUtil.binary(new byte[]{startFlag}, 16));
if (startFlag != (byte) 0x68) {
throw new DecoderException("Invalid start flag: " + startFlag);
}
// 读取数据长度
byte dataLength = in.readByte();
if (in.readableBytes() < dataLength + 2) { // 消息体 + 帧校验域长度
in.resetReaderIndex();
return; // 数据不足，等待更多字节
}
// 读取其他字段
short serialNumber = in.readShort(); // 序列号域
log.info("serialNumber: {}", BytesUtil.printHexBinary(new byte[]{(byte) serialNumber}));
byte encryptFlag = in.readByte(); // 加密标志
log.info("encryptFlag: {}", BytesUtil.printHexBinary(new byte[]{encryptFlag}));
byte frameType = in.readByte(); // 帧类型标志
log.info("frameType: {}", BytesUtil.printHexBinary(new byte[]{frameType}));
// 读取消息体
byte[] messageBody = new byte[dataLength - 4]; // 消息体长度 = 数据长度 - 固定字段长度
log.info("messageBody: {}", BytesUtil.printHexBinary(messageBody));
in.readBytes(messageBody);
// 读取帧校验域
short receivedCrc = in.readShort();
// 计算 CRC
short calculatedCrc = CRC16Util.calculateCrc(serialNumber, encryptFlag, frameType, messageBody);
if (calculatedCrc != receivedCrc) {
throw new DecoderException("CRC check failed. Expected: " + receivedCrc + ", Calculated: " + calculatedCrc);
}
// 构造消息对象并传递给下一个处理器
ProtocolMessage message = new ProtocolMessage(startFlag, dataLength, serialNumber, encryptFlag, frameType, messageBody, receivedCrc);
out.add(message);
}
public static int bcdToDecimal(byte bcd1, byte bcd2) {
// BCD码转换为十进制
int high = (bcd1 >> 4) & 0x0F; // 高4位
int low = bcd1 & 0x0F; // 低4位
int high2 = (bcd2 >> 4) & 0x0F;
int low2 = bcd2 & 0x0F;
// 将BCD码拼接成十进制数
return (high * 1000) + (low * 100) + (high2 * 10) + low2;
}
// 自定义消息类
public static class ProtocolMessage {
private final byte startFlag;
private final byte dataLength;
private final short serialNumber;
private final byte encryptFlag;
private final byte frameType;
private final byte[] messageBody;
private final short crc;
public ProtocolMessage(byte startFlag, byte dataLength, short serialNumber, byte encryptFlag, byte frameType, byte[] messageBody, short crc) {
this.startFlag = startFlag;
this.dataLength = dataLength;
this.serialNumber = serialNumber;
this.encryptFlag = encryptFlag;
this.frameType = frameType;
this.messageBody = messageBody;
this.crc = crc;
}
// Getter methods...
public byte getStartFlag() { return startFlag; }
public byte getDataLength() { return dataLength; }
public short getSerialNumber() { return serialNumber; }
public byte getEncryptFlag() { return encryptFlag; }
public byte getFrameType() { return frameType; }
public byte[] getMessageBody() { return messageBody; }
public short getCrc() { return crc; }
}
}

141
jsowell-netty/src/main/java/com/jsowell/netty/decoder/YouDianProtocolDecoder.java Normal file
View File

@@ -0,0 +1,141 @@
package com.jsowell.netty.decoder;
import com.jsowell.common.constant.Constants;
import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.ByteToMessageDecoder;
import lombok.extern.slf4j.Slf4j;
import java.nio.charset.StandardCharsets;
import java.util.List;
/**
* 友电协议解码器
*/
@Slf4j
public class YouDianProtocolDecoder extends ByteToMessageDecoder {
private static final int HEADER_LENGTH_DNY = 3; // "DNY" 包头的长度
private static final int HEADER_LENGTH_68 = 1; // 68 包头的长度
// 构造函数，初始化起始标志
public YouDianProtocolDecoder() {}
@Override
protected void decode(ChannelHandlerContext ctx, ByteBuf buffer, List<Object> out) throws Exception {
// log.info("StartAndLengthFieldFrameDecoder.decode");
// 记录包头开始的index
int beginReader;
// 循环查找包头
while (true) {
if (buffer.readableBytes() < Math.min(HEADER_LENGTH_DNY, HEADER_LENGTH_68)) {
return; // 数据长度不足，等待更多数据
}
// 获取包头开始的index
beginReader = buffer.readerIndex();
buffer.markReaderIndex();
// 判断是否为DNY包头或68包头
if (isStartOfDnyHeader(buffer, beginReader) || isStartOf68Header(buffer, beginReader)) {
break; // 读到了协议的开始标志，结束while循环
}
// 未读到包头，略过一个字节
buffer.resetReaderIndex();
buffer.readByte();
}
// 检查包头是否是 "DNY"
if (buffer.readableBytes() >= HEADER_LENGTH_DNY) {
byte[] headerBytes = new byte[HEADER_LENGTH_DNY];
buffer.getBytes(beginReader, headerBytes, 0, HEADER_LENGTH_DNY);
String header = new String(headerBytes, StandardCharsets.UTF_8);
// log.info("检查包头是否是DNY, header:{}", header);
if (Constants.EBIKE_HEADER.equals(header)) {
// 处理 DNY 协议
decodeDnyMessage(buffer, out, beginReader);
return;
}
}
// 检查包头是否是 68 协议
if (buffer.readableBytes() >= HEADER_LENGTH_68) {
if (buffer.getUnsignedByte(beginReader) == 0x68) {
// 处理 68 协议
decode68Message(buffer, out, beginReader);
return;
}
}
// 未知协议，还原读指针
buffer.resetReaderIndex();
}
// 判断是否为DNY包头
private boolean isStartOfDnyHeader(ByteBuf buffer, int beginReader) {
if (buffer.readableBytes() >= HEADER_LENGTH_DNY) {
byte[] headerBytes = new byte[HEADER_LENGTH_DNY];
buffer.getBytes(beginReader, headerBytes, 0, HEADER_LENGTH_DNY);
String header = new String(headerBytes, StandardCharsets.UTF_8);
return Constants.EBIKE_HEADER.equals(header);
}
return false;
}
// 判断是否为68包头
private boolean isStartOf68Header(ByteBuf buffer, int beginReader) {
if (buffer.readableBytes() >= HEADER_LENGTH_68) {
return buffer.getUnsignedByte(beginReader) == 0x68;
}
return false;
}
// 处理68协议消息
private void decode68Message(ByteBuf buffer, List<Object> out, int beginReader) {
// 检查剩余数据是否足够
if (buffer.readableBytes() < HEADER_LENGTH_68 + 1 + 2) {
buffer.readerIndex(beginReader);
return;
}
// 获取消息长度
int length = buffer.getUnsignedByte(beginReader + HEADER_LENGTH_68);
// 检查剩余数据是否足够
if (buffer.readableBytes() < HEADER_LENGTH_68 + 1 + length + 2) {
buffer.readerIndex(beginReader);
return;
}
// 读取 data 数据 最后+2是帧校验域长度
ByteBuf frame = buffer.retainedSlice(beginReader, HEADER_LENGTH_68 + 1 + length + 2);
buffer.readerIndex(beginReader + HEADER_LENGTH_68 + 1 + length + 2);
out.add(frame);
}
// 处理DNY协议消息
private void decodeDnyMessage(ByteBuf buffer, List<Object> out, int beginReader) {
// 检查剩余数据是否足够
if (buffer.readableBytes() < HEADER_LENGTH_DNY + 1) {
buffer.readerIndex(beginReader);
return;
}
// 获取消息长度
int length = buffer.getUnsignedByte(beginReader + HEADER_LENGTH_DNY);
// log.info("获取消息长度, length:{}", length);
// 检查剩余数据是否足够
if (buffer.readableBytes() < HEADER_LENGTH_DNY + 1 + length) {
buffer.readerIndex(beginReader);
return;
}
// 读取 data 数据
ByteBuf frame = buffer.retainedSlice(beginReader, HEADER_LENGTH_DNY + length + 2);
buffer.readerIndex(beginReader + HEADER_LENGTH_DNY + length + 2);
out.add(frame);
}
}

144
jsowell-netty/src/main/java/com/jsowell/netty/decoder/YunKuaiChongDecoder.java Normal file
View File

@@ -0,0 +1,144 @@
package com.jsowell.netty.decoder;
import com.jsowell.common.constant.Constants;
import com.jsowell.common.core.domain.ykc.YKCDataProtocol;
import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.ByteToMessageDecoder;
import lombok.extern.slf4j.Slf4j;
import java.nio.charset.StandardCharsets;
import java.util.List;
@Slf4j
public class YunKuaiChongDecoder extends ByteToMessageDecoder {
private static final int HEADER_LENGTH_DNY = 3; // "DNY" 包头的长度
private static final int HEADER_LENGTH_68 = 1; // 68 包头的长度
// 构造函数，初始化起始标志
public YunKuaiChongDecoder() {}
@Override
protected void decode(ChannelHandlerContext ctx, ByteBuf buffer, List<Object> out) throws Exception {
// log.info("StartAndLengthFieldFrameDecoder.decode");
// 记录包头开始的index
int beginReader;
// 循环查找包头
while (true) {
if (buffer.readableBytes() < Math.min(HEADER_LENGTH_DNY, HEADER_LENGTH_68)) {
return; // 数据长度不足，等待更多数据
}
// 获取包头开始的index
beginReader = buffer.readerIndex();
buffer.markReaderIndex();
// 判断是否为DNY包头或68包头
if (isStartOfDnyHeader(buffer, beginReader) || isStartOf68Header(buffer, beginReader)) {
break; // 读到了协议的开始标志，结束while循环
}
// 未读到包头，略过一个字节
buffer.resetReaderIndex();
buffer.readByte();
}
// 检查包头是否是 "DNY"
if (buffer.readableBytes() >= HEADER_LENGTH_DNY) {
byte[] headerBytes = new byte[HEADER_LENGTH_DNY];
buffer.getBytes(beginReader, headerBytes, 0, HEADER_LENGTH_DNY);
String header = new String(headerBytes, StandardCharsets.UTF_8);
// log.info("检查包头是否是DNY, header:{}", header);
if (Constants.EBIKE_HEADER.equals(header)) {
// 处理 DNY 协议
decodeDnyMessage(buffer, out, beginReader);
return;
}
}
// 检查包头是否是 68 协议
if (buffer.readableBytes() >= HEADER_LENGTH_68) {
if (buffer.getUnsignedByte(beginReader) == 0x68) {
// 处理 68 协议
decode68Message(buffer, out, beginReader);
return;
}
}
// 未知协议，还原读指针
buffer.resetReaderIndex();
}
// 判断是否为DNY包头
private boolean isStartOfDnyHeader(ByteBuf buffer, int beginReader) {
if (buffer.readableBytes() >= HEADER_LENGTH_DNY) {
byte[] headerBytes = new byte[HEADER_LENGTH_DNY];
buffer.getBytes(beginReader, headerBytes, 0, HEADER_LENGTH_DNY);
String header = new String(headerBytes, StandardCharsets.UTF_8);
return Constants.EBIKE_HEADER.equals(header);
}
return false;
}
// 判断是否为68包头
private boolean isStartOf68Header(ByteBuf buffer, int beginReader) {
if (buffer.readableBytes() >= HEADER_LENGTH_68) {
return buffer.getUnsignedByte(beginReader) == 0x68;
}
return false;
}
// 处理68协议消息
private void decode68Message(ByteBuf buffer, List<Object> out, int beginReader) {
// 检查剩余数据是否足够
if (buffer.readableBytes() < HEADER_LENGTH_68 + 1 + 2) {
buffer.readerIndex(beginReader);
return;
}
// 获取消息长度
int length = buffer.getUnsignedByte(beginReader + HEADER_LENGTH_68);
// 检查剩余数据是否足够
if (buffer.readableBytes() < HEADER_LENGTH_68 + 1 + length + 2) {
buffer.readerIndex(beginReader);
return;
}
// 读取 data 数据 最后+2是帧校验域长度
ByteBuf frame = buffer.retainedSlice(beginReader, HEADER_LENGTH_68 + 1 + length + 2);
buffer.readerIndex(beginReader + HEADER_LENGTH_68 + 1 + length + 2);
// 转为YKCDataProtocol对象
byte[] bytes = new byte[HEADER_LENGTH_68 + 1 + length + 2];
frame.readBytes(bytes);
YKCDataProtocol ykcDataProtocol = new YKCDataProtocol(bytes);
out.add(ykcDataProtocol);
}
// 处理DNY协议消息
private void decodeDnyMessage(ByteBuf buffer, List<Object> out, int beginReader) {
// 检查剩余数据是否足够
if (buffer.readableBytes() < HEADER_LENGTH_DNY + 1) {
buffer.readerIndex(beginReader);
return;
}
// 获取消息长度
int length = buffer.getUnsignedByte(beginReader + HEADER_LENGTH_DNY);
// log.info("获取消息长度, length:{}", length);
// 检查剩余数据是否足够
if (buffer.readableBytes() < HEADER_LENGTH_DNY + 1 + length) {
buffer.readerIndex(beginReader);
return;
}
// 读取 data 数据
ByteBuf frame = buffer.retainedSlice(beginReader, HEADER_LENGTH_DNY + length + 2);
buffer.readerIndex(beginReader + HEADER_LENGTH_DNY + length + 2);
out.add(frame);
}
}

4
jsowell-netty/src/main/java/com/jsowell/netty/server/electricbicycles/ElectricBicyclesServerChannelInitializer.java
View File

@@ -1,6 +1,6 @@
package com.jsowell.netty.server.electricbicycles;
import com.jsowell.netty.decoder.StartAndLengthFieldFrameDecoder;
import com.jsowell.netty.decoder.YouDianProtocolDecoder;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.socket.SocketChannel;
@@ -20,7 +20,7 @@ public class ElectricBicyclesServerChannelInitializer extends ChannelInitializer
@Override
protected void initChannel(SocketChannel channel) throws Exception {
ChannelPipeline pipeline = channel.pipeline();
pipeline.addLast("frameDecoder", new StartAndLengthFieldFrameDecoder());
pipeline.addLast("frameDecoder", new YouDianProtocolDecoder());
// pipeline.addLast("decoder", new MessageDecode());
// pipeline.addLast("encoder", new MessageEncode());
pipeline.addLast("decoder", new ByteArrayDecoder());

5
jsowell-netty/src/main/java/com/jsowell/netty/server/yunkuaichong/NettyServerChannelInitializer.java
View File

@@ -1,6 +1,6 @@
package com.jsowell.netty.server.yunkuaichong;
import com.jsowell.netty.decoder.CustomLengthFieldBasedFrameDecoder;
import com.jsowell.netty.decoder.YunKuaiChongDecoder;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.socket.SocketChannel;
@@ -22,7 +22,8 @@ public class NettyServerChannelInitializer extends ChannelInitializer<SocketChan
ChannelPipeline pipeline = channel.pipeline();
// pipeline.addLast("frameDecoder",new CustomDecoder());
// pipeline.addLast("frameDecoder", new StartAndLengthFieldFrameDecoder());
pipeline.addLast("frameDecoder", new CustomLengthFieldBasedFrameDecoder());
pipeline.addLast("frameDecoder",new YunKuaiChongDecoder());
// pipeline.addLast("frameDecoder", new YkcProtocolDecoder());
pipeline.addLast("decoder", new ByteArrayDecoder());
pipeline.addLast("encoder", new ByteArrayDecoder());
//读超时时间设置为10s，0表示不监控

75
jsowell-netty/src/main/java/com/jsowell/netty/server/yunkuaichong/NettyServerHandler.java
View File

@@ -1,6 +1,7 @@
package com.jsowell.netty.server.yunkuaichong;
import com.google.common.collect.Lists;
import com.jsowell.common.core.domain.ykc.YKCDataProtocol;
import com.jsowell.common.core.domain.ykc.YKCFrameTypeCode;
import com.jsowell.common.enums.ykc.PileChannelEntity;
import com.jsowell.common.util.BytesUtil;
@@ -66,14 +67,72 @@ public class NettyServerHandler extends ChannelInboundHandlerAdapter {
/**
* 有客户端发消息会触发此函数
*/
// @Override
// public void channelRead(ChannelHandlerContext ctx, Object message) throws Exception {
// // log.info("channelRead-aClass:{}", message.getClass());
// // log.info("加载客户端报文channelRead=== channelId:" + ctx.channel().id() + ", msg:" + message);
// // 下面可以解析数据，保存数据，生成返回报文，将需要返回报文写入write函数
// YKCDataProtocol ykcDataProtocol = (YKCDataProtocol) message;
// byte[] msg = ykcDataProtocol.getBytes();
//
// // 获取帧类型
// byte[] frameTypeBytes = BytesUtil.copyBytes(msg, 5, 1);
// String frameType = YKCUtils.frameType2Str(frameTypeBytes);
//
// // 判断该帧类型是否为某请求帧的应答帧
// String requestFrameType = YKCFrameTypeCode.PileAnswersRelation.getRequestFrameType(frameType);
// // log.info("同步获取响应数据-判断该帧类型是否为某请求帧的应答帧, frameType:{}, requestFrameType:{}", frameType, requestFrameType);
// if (StringUtils.isNotBlank(requestFrameType)) {
// // 根据请求id，在集合中找到与外部线程通信的SyncPromise对象
// String msgId = ctx.channel().id().toString() + "_" + requestFrameType;
// // log.info("同步获取响应数据-收到消息, msgId:{}", msgId);
// SyncPromise syncPromise = RpcUtil.getSyncPromiseMap().get(msgId);
// if(syncPromise != null) {
// // 设置响应结果
// syncPromise.setRpcResult(msg);
// // 唤醒外部线程
// // log.info("同步获取响应数据-唤醒外部线程, SyncPromise:{}", JSON.toJSONString(syncPromise));
// syncPromise.wake();
// }
// }
//
// // 获取序列号域
// int serialNumber = BytesUtil.bytesToIntLittle(BytesUtil.copyBytes(msg, 2, 2));
//
// // 获取channel
// Channel channel = ctx.channel();
//
// // 心跳包0x03日志太多，造成日志文件过大，改为不打印
// if (!CollectionUtils.containsAny(notPrintFrameTypeList, frameType)) {
// log.info("【<<<<<平台收到消息<<<<<】channel:{}, 帧类型:{}, 帧名称:{}, 序列号域:{}, 报文:{}",
// channel.id(), frameType, YKCFrameTypeCode.getFrameTypeStr(frameType), serialNumber,
// BytesUtil.binary(msg, 16));
// }
//
// // 处理数据
// byte[] response = ykcService.process(msg, ctx);
// if (Objects.nonNull(response)) {
// // 响应客户端
// ByteBuf buffer = ctx.alloc().buffer().writeBytes(response);
// this.channelWrite(channel.id(), buffer);
// if (!CollectionUtils.containsAny(notPrintFrameTypeList, frameType)) {
// // 应答帧类型
// byte[] responseFrameTypeBytes = YKCFrameTypeCode.PlatformAnswersRelation.getResponseFrameTypeBytes(frameTypeBytes);
// String responseFrameType = YKCUtils.frameType2Str(responseFrameTypeBytes);
// log.info("【>>>>>平台响应消息>>>>>】channel:{}, 响应帧类型:{}, 响应帧名称:{}, 原帧类型:{}, 原帧名称:{}, 序列号域:{}, response:{}",
// channel.id(), responseFrameType, YKCFrameTypeCode.getFrameTypeStr(responseFrameType),
// frameType, YKCFrameTypeCode.getFrameTypeStr(frameType), serialNumber,
// BytesUtil.binary(response, 16));
// }
// }
// }
@Override
public void channelRead(ChannelHandlerContext ctx, Object message) throws Exception {
// log.info("加载客户端报文channelRead=== channelId:" + ctx.channel().id() + ", msg:" + message);
// 下面可以解析数据，保存数据，生成返回报文，将需要返回报文写入write函数
byte[] msg = (byte[]) message;
YKCDataProtocol ykcDataProtocol = (YKCDataProtocol) message;
// 获取帧类型
byte[] frameTypeBytes = BytesUtil.copyBytes(msg, 5, 1);
byte[] frameTypeBytes = ykcDataProtocol.getFrameType();
String frameType = YKCUtils.frameType2Str(frameTypeBytes);
// 判断该帧类型是否为某请求帧的应答帧
@@ -86,7 +145,7 @@ public class NettyServerHandler extends ChannelInboundHandlerAdapter {
SyncPromise syncPromise = RpcUtil.getSyncPromiseMap().get(msgId);
if(syncPromise != null) {
// 设置响应结果
syncPromise.setRpcResult(msg);
syncPromise.setRpcResult(ykcDataProtocol.getBytes());
// 唤醒外部线程
// log.info("同步获取响应数据-唤醒外部线程, SyncPromise:{}", JSON.toJSONString(syncPromise));
syncPromise.wake();
@@ -94,7 +153,7 @@ public class NettyServerHandler extends ChannelInboundHandlerAdapter {
}
// 获取序列号域
int serialNumber = BytesUtil.bytesToIntLittle(BytesUtil.copyBytes(msg, 2, 2));
int serialNumber = BytesUtil.bytesToIntLittle(ykcDataProtocol.getSerialNumber());
// 获取channel
Channel channel = ctx.channel();
@@ -103,11 +162,11 @@ public class NettyServerHandler extends ChannelInboundHandlerAdapter {
if (!CollectionUtils.containsAny(notPrintFrameTypeList, frameType)) {
log.info("【<<<<<平台收到消息<<<<<】channel:{}, 帧类型:{}, 帧名称:{}, 序列号域:{}, 报文:{}",
channel.id(), frameType, YKCFrameTypeCode.getFrameTypeStr(frameType), serialNumber,
BytesUtil.binary(msg, 16));
BytesUtil.binary(ykcDataProtocol.getBytes(), 16));
}
// 处理数据
byte[] response = ykcService.process(msg, ctx);
byte[] response = ykcService.process(ykcDataProtocol, ctx);
if (Objects.nonNull(response)) {
// 响应客户端
ByteBuf buffer = ctx.alloc().buffer().writeBytes(response);

4
jsowell-netty/src/main/java/com/jsowell/netty/service/yunkuaichong/YKCBusinessService.java
View File

@@ -1,6 +1,6 @@
package com.jsowell.netty.service.yunkuaichong;
import io.netty.channel.Channel;
import com.jsowell.common.core.domain.ykc.YKCDataProtocol;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelId;
@@ -18,6 +18,8 @@ public interface YKCBusinessService {
*/
byte[] process(byte[] msg, ChannelHandlerContext ctx);
byte[] process(YKCDataProtocol ykcDataProtocol, ChannelHandlerContext ctx);
/**
* 桩退出
* @param channelId channelId

14
jsowell-netty/src/main/java/com/jsowell/netty/service/yunkuaichong/impl/YKCBusinessServiceImpl.java
View File

@@ -49,6 +49,20 @@ public class YKCBusinessServiceImpl implements YKCBusinessService {
return invokeStrategy.supplyProcess(ykcDataProtocol, ctx);
}
@Override
public byte[] process(YKCDataProtocol ykcDataProtocol, ChannelHandlerContext ctx) {
if (!YKCUtils.checkMsg(ykcDataProtocol)) {
// 校验不通过，丢弃消息
return null;
}
// 获取帧类型
String frameType = YKCUtils.frameType2Str(ykcDataProtocol.getFrameType());
// 获取业务处理handler
AbstractYkcHandler invokeStrategy = YKCOperateFactory.getInvokeStrategy(frameType);
// AbstractYkcHandlerV2 invokeStrategy = ykcOperateFactoryV2.getInvokeStrategy(frameType);
return invokeStrategy.supplyProcess(ykcDataProtocol, ctx);
}
@Override
public void exit(ChannelId channelId) {
// 获取桩编号