tiennm99/zfoo
1
0
Fork 0
mirror of https://github.com/tiennm99/zfoo.git synced 2026-08-06 16:25:28 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge remote-tracking branch 'origin/main'

Browse Source
This commit is contained in:
godotg
2023-04-08 16:54:21 +08:00
parent ef2847aa64 cb6c72a78b
commit 84b8d20e95
5 changed files with 55 additions and 163 deletions
Download Patch File Download Diff File Expand all files Collapse all files
net/src/main/java/com/zfoo/net/handler/codec/http/HttpCodecHandler.java
+31 -49
View File
@@ -35,8 +35,6 @@ import java.util.function.Function;
*/
public class HttpCodecHandler extends MessageToMessageCodec<FullHttpRequest, EncodedPacketInfo> {
private static final Logger logger = LoggerFactory.getLogger(HttpCodecHandler.class);
private final Function<FullHttpRequest, DecodedPacketInfo> uriResolver;
public HttpCodecHandler(Function<FullHttpRequest, DecodedPacketInfo> uriResolver) {
@@ -46,67 +44,51 @@ public class HttpCodecHandler extends MessageToMessageCodec<FullHttpRequest, Enc
@Override
protected void decode(ChannelHandlerContext channelHandlerContext, FullHttpRequest fullHttpRequest, List<Object> list) {
try {
var decodedPacketInfo = uriResolver.apply(fullHttpRequest);
list.add(decodedPacketInfo);
} catch (Exception e) {
logger.error("exception异常", e);
throw e;
} catch (Throwable t) {
logger.error("throwable错误", t);
throw t;
}
var decodedPacketInfo = uriResolver.apply(fullHttpRequest);
list.add(decodedPacketInfo);
}
@Override
protected void encode(ChannelHandlerContext channelHandlerContext, EncodedPacketInfo out, List<Object> list) {
try {
var packet = (IPacket) out.getPacket();
var attachment = (HttpAttachment) out.getAttachment();
var packet = (IPacket) out.getPacket();
var attachment = (HttpAttachment) out.getAttachment();
var protocolVersion = attachment.getFullHttpRequest().protocolVersion();
var httpResponseStatus = attachment.getHttpResponseStatus();
if (packet.protocolId() == Message.PROTOCOL_ID) {
var message = (Message) packet;
if (message.fail()) {
httpResponseStatus = HttpResponseStatus.BAD_REQUEST;
}
var protocolVersion = attachment.getFullHttpRequest().protocolVersion();
var httpResponseStatus = attachment.getHttpResponseStatus();
if (packet.protocolId() == Message.PROTOCOL_ID) {
var message = (Message) packet;
if (message.fail()) {
httpResponseStatus = HttpResponseStatus.BAD_REQUEST;
}
if (StringUtils.isEmpty(message.getMessage())) {
var fullHttpResponse = new DefaultFullHttpResponse(protocolVersion, httpResponseStatus);
fullHttpResponse.headers().set(HttpHeaderNames.CONTENT_TYPE, HttpHeaderValues.TEXT_PLAIN);
fullHttpResponse.headers().set(HttpHeaderNames.CONTENT_LENGTH, 0);
fullHttpResponse.headers().set(HttpHeaderNames.CONNECTION, HttpHeaderValues.KEEP_ALIVE);
list.add(fullHttpResponse);
} else {
var byteBuf = channelHandlerContext.alloc().ioBuffer();
byteBuf.writeCharSequence(message.getMessage(), StringUtils.DEFAULT_CHARSET);
var fullHttpResponse = new DefaultFullHttpResponse(protocolVersion, httpResponseStatus, byteBuf);
fullHttpResponse.headers().set(HttpHeaderNames.CONTENT_TYPE, HttpHeaderValues.TEXT_PLAIN);
fullHttpResponse.headers().set(HttpHeaderNames.CONTENT_LENGTH, byteBuf.readableBytes());
fullHttpResponse.headers().set(HttpHeaderNames.CONNECTION, HttpHeaderValues.KEEP_ALIVE);
list.add(fullHttpResponse);
}
if (StringUtils.isEmpty(message.getMessage())) {
var fullHttpResponse = new DefaultFullHttpResponse(protocolVersion, httpResponseStatus);
fullHttpResponse.headers().set(HttpHeaderNames.CONTENT_TYPE, HttpHeaderValues.TEXT_PLAIN);
fullHttpResponse.headers().set(HttpHeaderNames.CONTENT_LENGTH, 0);
fullHttpResponse.headers().set(HttpHeaderNames.CONNECTION, HttpHeaderValues.KEEP_ALIVE);
list.add(fullHttpResponse);
} else {
var byteBuf = channelHandlerContext.alloc().ioBuffer();
var jsonStr = JsonUtils.object2String(packet);
byteBuf.writeBytes(StringUtils.bytes(jsonStr));
byteBuf.writeCharSequence(message.getMessage(), StringUtils.DEFAULT_CHARSET);
var fullHttpResponse = new DefaultFullHttpResponse(protocolVersion, httpResponseStatus, byteBuf);
fullHttpResponse.headers().set(HttpHeaderNames.CONTENT_TYPE, HttpHeaderValues.APPLICATION_JSON);
fullHttpResponse.headers().set(HttpHeaderNames.CONTENT_TYPE, HttpHeaderValues.TEXT_PLAIN);
fullHttpResponse.headers().set(HttpHeaderNames.CONTENT_LENGTH, byteBuf.readableBytes());
fullHttpResponse.headers().set(HttpHeaderNames.CONNECTION, HttpHeaderValues.KEEP_ALIVE);
// fullHttpResponse.headers().set(HttpHeaderNames.TRANSFER_ENCODING, HttpHeaderValues.CHUNKED);
list.add(fullHttpResponse);
}
} catch (Exception e) {
logger.error("[{}]编码exception异常", JsonUtils.object2String(out), e);
throw e;
} catch (Throwable t) {
logger.error("[{}]编码throwable错误", JsonUtils.object2String(out), t);
throw t;
} else {
var byteBuf = channelHandlerContext.alloc().ioBuffer();
var jsonStr = JsonUtils.object2String(packet);
byteBuf.writeBytes(StringUtils.bytes(jsonStr));
var fullHttpResponse = new DefaultFullHttpResponse(protocolVersion, httpResponseStatus, byteBuf);
fullHttpResponse.headers().set(HttpHeaderNames.CONTENT_TYPE, HttpHeaderValues.APPLICATION_JSON);
fullHttpResponse.headers().set(HttpHeaderNames.CONTENT_LENGTH, byteBuf.readableBytes());
fullHttpResponse.headers().set(HttpHeaderNames.CONNECTION, HttpHeaderValues.KEEP_ALIVE);
// fullHttpResponse.headers().set(HttpHeaderNames.TRANSFER_ENCODING, HttpHeaderValues.CHUNKED);
list.add(fullHttpResponse);
}
}
}
net/src/main/java/com/zfoo/net/handler/codec/jprotobuf/JProtobufTcpCodecHandler.java
+4 -31
View File
@@ -44,9 +44,6 @@ import java.util.List;
*/
public class JProtobufTcpCodecHandler extends ByteToMessageCodec<EncodedPacketInfo> {
private static final Logger logger = LoggerFactory.getLogger(JProtobufTcpCodecHandler.class);
@Override
protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws IOException {
// 不够读一个int
@@ -67,33 +64,14 @@ public class JProtobufTcpCodecHandler extends ByteToMessageCodec<EncodedPacketIn
return;
}
ByteBuf tmpByteBuf = null;
try {
tmpByteBuf = in.readRetainedSlice(length);
DecodedPacketInfo packetInfo = read(tmpByteBuf);
out.add(packetInfo);
} catch (Exception e) {
logger.error("decode exception {}", SessionUtils.sessionSimpleInfo(ctx), e);
throw e;
} catch (Throwable t) {
logger.error("decode throwable {}", SessionUtils.sessionSimpleInfo(ctx), t);
throw t;
} finally {
ReferenceCountUtil.release(tmpByteBuf);
}
var sliceByteBuf = in.readSlice(length);
var packetInfo = read(sliceByteBuf);
out.add(packetInfo);
}
@Override
protected void encode(ChannelHandlerContext ctx, EncodedPacketInfo packetInfo, ByteBuf out) throws IOException {
try {
write(out, packetInfo.getPacket(), packetInfo.getAttachment());
} catch (Exception e) {
logger.error("[{}] encode exception {}", SessionUtils.sessionSimpleInfo(ctx), packetInfo.getPacket().getClass().getSimpleName(), e);
throw e;
} catch (Throwable t) {
logger.error("[{}] encode throwable {}", SessionUtils.sessionSimpleInfo(ctx), packetInfo.getPacket().getClass().getSimpleName(), t);
throw t;
}
write(out, packetInfo.getPacket(), packetInfo.getAttachment());
}
public static DecodedPacketInfo read(ByteBuf buffer) throws IOException {
@@ -110,11 +88,6 @@ public class JProtobufTcpCodecHandler extends ByteToMessageCodec<EncodedPacketIn
}
public void write(ByteBuf buffer, IPacket packet, IAttachment attachment) throws IOException {
if (packet == null) {
logger.error("packet is null and can not be sent.");
return;
}
// 写入protobuf协议
var protobufCodec = (Codec<IPacket>) ProtobufProxy.create(packet.getClass());
byte[] bytes = protobufCodec.encode(packet);
net/src/main/java/com/zfoo/net/handler/codec/tcp/TcpCodecHandler.java
+5 -26
View File
@@ -38,8 +38,6 @@ import java.util.List;
*/
public class TcpCodecHandler extends ByteToMessageCodec<EncodedPacketInfo> {
private static final Logger logger = LoggerFactory.getLogger(TcpCodecHandler.class);
@Override
protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) {
// 不够读一个int
@@ -60,34 +58,15 @@ public class TcpCodecHandler extends ByteToMessageCodec<EncodedPacketInfo> {
return;
}
ByteBuf tmpByteBuf = null;
try {
// readRetainedSlice和byte[]数组相比，readRetainedSlice减少了垃圾回收
tmpByteBuf = in.readRetainedSlice(length);
DecodedPacketInfo packetInfo = NetContext.getPacketService().read(tmpByteBuf);
out.add(packetInfo);
} catch (Exception e) {
logger.error("decode exception {}", SessionUtils.sessionSimpleInfo(ctx), e);
throw e;
} catch (Throwable t) {
logger.error("decode throwable {}", SessionUtils.sessionSimpleInfo(ctx), t);
throw t;
} finally {
ReferenceCountUtil.release(tmpByteBuf);
}
// readSlice和byte[]数组相比，readSlice减少了垃圾回收
var sliceByteBuf = in.readSlice(length);
var packetInfo = NetContext.getPacketService().read(sliceByteBuf);
out.add(packetInfo);
}
@Override
protected void encode(ChannelHandlerContext ctx, EncodedPacketInfo packetInfo, ByteBuf out) {
try {
NetContext.getPacketService().write(out, packetInfo.getPacket(), packetInfo.getAttachment());
} catch (Exception e) {
logger.error("[{}] encode exception {}", SessionUtils.sessionSimpleInfo(ctx), packetInfo.getPacket().getClass().getSimpleName(), e);
throw e;
} catch (Throwable t) {
logger.error("[{}] encode throwable {}", SessionUtils.sessionSimpleInfo(ctx), packetInfo.getPacket().getClass().getSimpleName(), t);
throw t;
}
NetContext.getPacketService().write(out, packetInfo.getPacket(), packetInfo.getAttachment());
}
}
net/src/main/java/com/zfoo/net/handler/codec/udp/UdpCodecHandler.java
+9 -30
View File
@@ -37,8 +37,6 @@ import java.util.List;
*/
public class UdpCodecHandler extends MessageToMessageCodec<DatagramPacket, EncodedPacketInfo> {
private static final Logger logger = LoggerFactory.getLogger(UdpCodecHandler.class);
@Override
protected void decode(ChannelHandlerContext channelHandlerContext, DatagramPacket datagramPacket, List<Object> list) {
ByteBuf in = datagramPacket.content();
@@ -62,38 +60,19 @@ public class UdpCodecHandler extends MessageToMessageCodec<DatagramPacket, Encod
return;
}
ByteBuf tmpByteBuf = null;
try {
tmpByteBuf = in.readRetainedSlice(length);
DecodedPacketInfo packetInfo = NetContext.getPacketService().read(tmpByteBuf);
var sender = datagramPacket.sender();
packetInfo.setAttachment(UdpAttachment.valueOf(sender.getHostString(), sender.getPort()));
list.add(packetInfo);
} catch (Exception e) {
logger.error("exception异常", e);
throw e;
} catch (Throwable t) {
logger.error("throwable错误", t);
throw t;
} finally {
ReferenceCountUtil.release(tmpByteBuf);
}
var sliceByteBuf = in.readSlice(length);
var packetInfo = NetContext.getPacketService().read(sliceByteBuf);
var sender = datagramPacket.sender();
packetInfo.setAttachment(UdpAttachment.valueOf(sender.getHostString(), sender.getPort()));
list.add(packetInfo);
}
@Override
protected void encode(ChannelHandlerContext channelHandlerContext, EncodedPacketInfo out, List<Object> list) {
try {
var byteBuf = channelHandlerContext.alloc().ioBuffer();
var udpAttachment = (UdpAttachment) out.getAttachment();
var byteBuf = channelHandlerContext.alloc().ioBuffer();
var udpAttachment = (UdpAttachment) out.getAttachment();
NetContext.getPacketService().write(byteBuf, out.getPacket(), out.getAttachment());
list.add(new DatagramPacket(byteBuf, new InetSocketAddress(udpAttachment.getHost(), udpAttachment.getPort())));
} catch (Exception e) {
logger.error("[{}]编码exception异常", JsonUtils.object2String(out), e);
throw e;
} catch (Throwable t) {
logger.error("[{}]编码throwable错误", JsonUtils.object2String(out), t);
throw t;
}
NetContext.getPacketService().write(byteBuf, out.getPacket(), out.getAttachment());
list.add(new DatagramPacket(byteBuf, new InetSocketAddress(udpAttachment.getHost(), udpAttachment.getPort())));
}
}
net/src/main/java/com/zfoo/net/handler/codec/websocket/WebSocketCodecHandler.java
+6 -27
View File
@@ -40,8 +40,6 @@ import java.util.List;
*/
public class WebSocketCodecHandler extends MessageToMessageCodec<WebSocketFrame, EncodedPacketInfo> {
private static final Logger logger = LoggerFactory.getLogger(WebSocketCodecHandler.class);
@Override
protected void decode(ChannelHandlerContext channelHandlerContext, WebSocketFrame webSocketFrame, List<Object> list) {
ByteBuf in = webSocketFrame.content();
@@ -65,36 +63,17 @@ public class WebSocketCodecHandler extends MessageToMessageCodec<WebSocketFrame,
return;
}
ByteBuf tmpByteBuf = null;
try {
tmpByteBuf = in.readRetainedSlice(length);
DecodedPacketInfo packetInfo = NetContext.getPacketService().read(tmpByteBuf);
list.add(packetInfo);
} catch (Exception e) {
logger.error("exception异常", e);
throw e;
} catch (Throwable t) {
logger.error("throwable错误", t);
throw t;
} finally {
ReferenceCountUtil.release(tmpByteBuf);
}
var sliceByteBuf = in.readSlice(length);
var packetInfo = NetContext.getPacketService().read(sliceByteBuf);
list.add(packetInfo);
}
@Override
protected void encode(ChannelHandlerContext channelHandlerContext, EncodedPacketInfo out, List<Object> list) {
try {
var byteBuf = channelHandlerContext.alloc().ioBuffer();
var byteBuf = channelHandlerContext.alloc().ioBuffer();
NetContext.getPacketService().write(byteBuf, out.getPacket(), out.getAttachment());
list.add(new BinaryWebSocketFrame(byteBuf));
} catch (Exception e) {
logger.error("[{}]编码exception异常", JsonUtils.object2String(out), e);
throw e;
} catch (Throwable t) {
logger.error("[{}]编码throwable错误", JsonUtils.object2String(out), t);
throw t;
}
NetContext.getPacketService().write(byteBuf, out.getPacket(), out.getAttachment());
list.add(new BinaryWebSocketFrame(byteBuf));
}