gitee/cim
1
0
Fork 0
mirror of https://gitee.com/farsunset/cim.git synced 2025-07-12 11:37:05 +08:00
Code Issues Packages Projects Releases Wiki Activity

优化解码socket数据

Browse Source
This commit is contained in:
xia jun 2024-06-11 12:53:52 +08:00
parent baa536f442
commit 7175585d37
2 changed files with 28 additions and 11 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
cim-server-sdk/src/main/java/com/farsunset/cim/auth/AuthHandler.java
View File

@ -40,7 +40,7 @@ public class AuthHandler extends ChannelInboundHandlerAdapter {
/* /*
* 鉴权不通过，发送响应体并关闭链接 * 鉴权不通过，发送响应体并关闭链接
*/ */
if (!authPredicate.test(AuthPredicateInfo.of(request, ctx))) { if (authPredicate != null && ! authPredicate.test(AuthPredicateInfo.of(request, ctx))) {
ctx.channel().writeAndFlush(failedBody).addListener(ChannelFutureListener.CLOSE); ctx.channel().writeAndFlush(failedBody).addListener(ChannelFutureListener.CLOSE);
return; return;
} }

37
cim-server-sdk/src/main/java/com/farsunset/cim/coder/protobuf/AppMessageDecoder.java
View File

@ -31,6 +31,8 @@ import com.farsunset.cim.model.proto.SentBodyProto;
import io.netty.buffer.ByteBuf; import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext; import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.ByteToMessageDecoder; import io.netty.handler.codec.ByteToMessageDecoder;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.List; import java.util.List;
@ -39,21 +41,40 @@ import java.util.List;
*/ */
public class AppMessageDecoder extends ByteToMessageDecoder { public class AppMessageDecoder extends ByteToMessageDecoder {
protected final Logger logger = LoggerFactory.getLogger(getClass());
@Override @Override
protected void decode(ChannelHandlerContext context, ByteBuf buffer, List<Object> queue) throws Exception { protected void decode(ChannelHandlerContext context, ByteBuf buffer, List<Object> queue) throws Exception {
context.channel().attr(ChannelAttr.PING_COUNT).set(null); context.channel().attr(ChannelAttr.PING_COUNT).set(null);
buffer.markReaderIndex();
byte type = buffer.readByte();
/*
* 收到意外数据
*/
if (DataType.PONG.getValue() != type && type != DataType.SENT.getValue()){
/*
消费掉无效数据
*/
int length = buffer.readableBytes();
if (length > 0){
buffer.readBytes(new byte[length]);
}
logger.warn("收到意外的数据,type={},length ={}", type,length);
return;
}
/* /*
* 消息体不足3位，发生断包情况 * 消息体不足3位，发生断包情况
*/ */
if (buffer.readableBytes() < CIMConstant.DATA_HEADER_LENGTH) { if (buffer.readableBytes() < CIMConstant.DATA_HEADER_LENGTH) {
buffer.resetReaderIndex();
return; return;
} }
buffer.markReaderIndex();
byte type = buffer.readByte();
byte lv = buffer.readByte(); byte lv = buffer.readByte();
byte hv = buffer.readByte(); byte hv = buffer.readByte();
@ -72,18 +93,14 @@ public class AppMessageDecoder extends ByteToMessageDecoder {
buffer.readBytes(content); buffer.readBytes(content);
Object message = mappingMessageObject(content, type); Object message = mappingMessageObject(content, type);
queue.add(message); queue.add(message);
} }
private Object mappingMessageObject(byte[] data, byte type) throws com.google.protobuf.InvalidProtocolBufferException { private Object mappingMessageObject(byte[] data, byte type) throws com.google.protobuf.InvalidProtocolBufferException {
if (DataType.PONG.getValue() == type) {
return Pong.getInstance();
}
if (DataType.SENT.getValue() == type) { if (DataType.SENT.getValue() == type) {
SentBodyProto.Model bodyProto = SentBodyProto.Model.parseFrom(data); SentBodyProto.Model bodyProto = SentBodyProto.Model.parseFrom(data);
SentBody body = new SentBody(); SentBody body = new SentBody();
body.setData(bodyProto.getDataMap()); body.setData(bodyProto.getDataMap());
@ -93,7 +110,7 @@ public class AppMessageDecoder extends ByteToMessageDecoder {
return body; return body;
} }
throw new ReadInvalidTypeException(type); return Pong.getInstance();
} }
/** /**