update 电单车协议

jsowell-netty/src/main/java/com/jsowell/netty/decoder/ChargingPileDecoder.java

@@ -4,9 +4,11 @@ import com.jsowell.netty.domain.ChargingPileMessage;
 import io.netty.buffer.ByteBuf;
 import io.netty.channel.ChannelHandlerContext;
 import io.netty.handler.codec.ByteToMessageDecoder;
+import lombok.extern.slf4j.Slf4j;
 
 import java.util.List;
 
+@Slf4j
 public class ChargingPileDecoder extends ByteToMessageDecoder {
 
     private static final byte[] FRAME_HEADER = {'D', 'N', 'Y'}; // 包头为"DNY"
@@ -52,7 +54,7 @@ public class ChargingPileDecoder extends ByteToMessageDecoder {
             // 验证校验和
             short calculatedChecksum = calculateChecksum(in, length);
             if (checksum != calculatedChecksum) {
-                System.out.println("校验和错误，丢弃此帧");
+                log.info("校验和错误，丢弃此帧");
                 continue;
             }
 
@@ -63,10 +65,12 @@ public class ChargingPileDecoder extends ByteToMessageDecoder {
     }
 
     private boolean isValidHeader(byte[] header) {
+        log.info("isValidHeader header:{}", header);
         return header[0] == FRAME_HEADER[0] && header[1] == FRAME_HEADER[1] && header[2] == FRAME_HEADER[2];
     }
 
     private short calculateChecksum(ByteBuf buf, int length) {
+        log.info("calculateChecksum ByteBuf:{}, length:{}", buf.toString(), length);
         short sum = 0;
         for (int i = 0; i < length - 2; i++) {
             sum += buf.getByte(buf.readerIndex() - length + i) & 0xFF;

jsowell-netty/src/main/java/com/jsowell/netty/decoder/StartAndLengthFieldFrameDecoder.java

@@ -16,77 +16,6 @@ public class StartAndLengthFieldFrameDecoder extends ByteToMessageDecoder {
 	// 构造函数，初始化起始标志
 	public StartAndLengthFieldFrameDecoder() {}
 
-	// 起始标志
-	// private int HEAD_DATA;
-
-
-	// public StartAndLengthFieldFrameDecoder(int HEAD_DATA) {
-	// 	this.HEAD_DATA = HEAD_DATA;
-	// }
-
-	/**
-	 * <pre>
-	 * 协议开始的标准head_data，int类型，占据1个字节.
-	 * 表示数据的长度contentLength，int类型，占据1个字节.
-	 * </pre>
-	 */
-	// public final int BASE_LENGTH = 1 + 1;
-
-	// @Override
-	// protected void decode2(ChannelHandlerContext ctx, ByteBuf buffer, List<Object> out) throws Exception {
-	// 	// 可读长度必须大于基本长度
-	// 	if (buffer.readableBytes() <= BASE_LENGTH) {
-	// 		log.warn("可读字节数:{}小于基础长度:{}", buffer.readableBytes(), BASE_LENGTH);
-	// 		return;
-	// 	}
-	//
-	// 	// 记录包头开始的index
-	// 	int beginReader;
-	//
-	// 	while (true) {
-	// 		// 获取包头开始的index
-	// 		beginReader = buffer.readerIndex();
-	// 		// log.info("包头开始的index:{}", beginReader);
-	// 		// 标记包头开始的index
-	// 		buffer.markReaderIndex();
-	// 		// 读到了协议的开始标志，结束while循环
-	// 		if (buffer.getUnsignedByte(beginReader) == HEAD_DATA) {
-	// 			// log.info("读到了协议的开始标志，结束while循环 byte:{}, HEAD_DATA:{}", buffer.getUnsignedByte(beginReader), HEAD_DATA);
-	// 			break;
-	// 		}
-	//
-	// 		// 未读到包头，略过一个字节
-	// 		// 每次略过，一个字节，去读取，包头信息的开始标记
-	// 		buffer.resetReaderIndex();
-	// 		buffer.readByte();
-	//
-	// 		// 当略过，一个字节之后，
-	// 		// 数据包的长度，又变得不满足
-	// 		// 此时，应该结束。等待后面的数据到达
-	// 		if (buffer.readableBytes() < BASE_LENGTH) {
-	// 			log.debug("数据包的长度不满足 readableBytes:{}, BASE_LENGTH:{}", buffer.readableBytes(), BASE_LENGTH);
-	// 			return;
-	// 		}
-	// 	}
-	//
-	// 	// 消息的长度
-	// 	int length = buffer.getUnsignedByte(beginReader + 1);
-	// 	// 判断请求数据包数据是否到齐
-	// 	if (buffer.readableBytes() < length + 4) {
-	// 		// log.info("请求数据包数据没有到齐,还原读指针 readableBytes:{}, 消息的长度:{}", buffer.readableBytes(), length);
-	// 		// 还原读指针
-	// 		buffer.readerIndex(beginReader);
-	// 		return;
-	// 	}
-	//
-	// 	// 读取data数据
-	// 	byte[] data = new byte[length + 4];
-	// 	buffer.readBytes(data);
-	// 	ByteBuf frame = buffer.retainedSlice(beginReader, length + 4);
-	// 	buffer.readerIndex(beginReader + length + 4);
-	// 	out.add(frame);
-	// }
-
 	protected void decode(ChannelHandlerContext ctx, ByteBuf buffer, List<Object> out) throws Exception {
 		// 记录包头开始的index
 		int beginReader;
@@ -101,11 +30,6 @@ public class StartAndLengthFieldFrameDecoder extends ByteToMessageDecoder {
 			beginReader = buffer.readerIndex();
 			buffer.markReaderIndex();
 
-			// 读到了协议的开始标志，结束while循环
-			// if (buffer.getUnsignedByte(beginReader) == HEAD_DATA) {
-			// 	break;
-			// }
-
 			// 判断是否为DNY包头或68包头
 			if (isStartOfDnyHeader(buffer, beginReader) || isStartOf68Header(buffer, beginReader)) {
 				break; // 读到了协议的开始标志，结束while循环

jsowell-netty/src/main/java/com/jsowell/netty/decoder/StartAndLengthFieldFrameDecoder2.java

@@ -0,0 +1,117 @@
+package com.jsowell.netty.decoder;
+
+import com.jsowell.netty.domain.DnyMessage;
+import com.jsowell.netty.domain.Message68;
+import io.netty.buffer.ByteBuf;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.handler.codec.ByteToMessageDecoder;
+import lombok.extern.slf4j.Slf4j;
+
+import java.util.List;
+
+@Slf4j
+public class StartAndLengthFieldFrameDecoder2 extends ByteToMessageDecoder {
+    private static final int HEADER_LENGTH_DNY = 3; // "DNY" 包头的长度
+    private static final int HEADER_LENGTH_68 = 1; // 68 包头的长度
+    private static final int MAX_FRAME_LENGTH = 1024; // 最大帧长度，可以根据实际需求调整
+    private static final byte[] DNY_HEADER = {'D', 'N', 'Y'};
+
+    @Override
+    protected void decode(ChannelHandlerContext ctx, ByteBuf buffer, List<Object> out) throws Exception {
+        while (buffer.readableBytes() > 0) {
+            int headerIndex = findHeaderIndex(buffer);
+            if (headerIndex == -1) {
+                // 没有找到有效的包头，丢弃所有数据
+                buffer.skipBytes(buffer.readableBytes());
+                return;
+            }
+
+            buffer.readerIndex(headerIndex);
+
+            if (isDnyHeader(buffer)) {
+                if (decodeDnyMessage(buffer, out)) {
+                    return;
+                }
+            } else if (is68Header(buffer)) {
+                if (decode68Message(buffer, out)) {
+                    return;
+                }
+            } else {
+                // 未知协议，跳过一个字节
+                buffer.skipBytes(1);
+            }
+        }
+    }
+
+    private int findHeaderIndex(ByteBuf buffer) {
+        int dnyIndex = buffer.indexOf(buffer.readerIndex(), buffer.writerIndex(), DNY_HEADER[0]);
+        int index68 = buffer.indexOf(buffer.readerIndex(), buffer.writerIndex(), (byte) 0x68);
+
+        if (dnyIndex == -1 && index68 == -1) {
+            return -1;
+        } else if (dnyIndex == -1) {
+            return index68;
+        } else if (index68 == -1) {
+            return dnyIndex;
+        } else {
+            return Math.min(dnyIndex, index68);
+        }
+    }
+
+    private boolean isDnyHeader(ByteBuf buffer) {
+        return buffer.readableBytes() >= HEADER_LENGTH_DNY &&
+                buffer.getByte(buffer.readerIndex()) == 'D' &&
+                buffer.getByte(buffer.readerIndex() + 1) == 'N' &&
+                buffer.getByte(buffer.readerIndex() + 2) == 'Y';
+    }
+
+    private boolean is68Header(ByteBuf buffer) {
+        return buffer.readableBytes() >= HEADER_LENGTH_68 &&
+                buffer.getByte(buffer.readerIndex()) == 0x68;
+    }
+
+    private boolean decodeDnyMessage(ByteBuf buffer, List<Object> out) {
+        if (buffer.readableBytes() < HEADER_LENGTH_DNY + 2) {
+            return false; // 数据不足，等待更多数据
+        }
+
+        int lengthFieldIndex = buffer.readerIndex() + HEADER_LENGTH_DNY;
+        int length = buffer.getUnsignedShort(lengthFieldIndex);
+
+        if (length > MAX_FRAME_LENGTH) {
+            log.warn("DNY frame length exceeds maximum allowed: {}", length);
+            buffer.skipBytes(HEADER_LENGTH_DNY + 2);
+            return false;
+        }
+
+        if (buffer.readableBytes() < HEADER_LENGTH_DNY + 2 + length) {
+            return false; // 数据不足，等待更多数据
+        }
+
+        ByteBuf frame = buffer.readRetainedSlice(HEADER_LENGTH_DNY + 2 + length);
+        out.add(new DnyMessage(frame));
+        return true;
+    }
+
+    private boolean decode68Message(ByteBuf buffer, List<Object> out) {
+        if (buffer.readableBytes() < HEADER_LENGTH_68 + 1) {
+            return false; // 数据不足，等待更多数据
+        }
+
+        int length = buffer.getUnsignedByte(buffer.readerIndex() + HEADER_LENGTH_68);
+
+        if (length > MAX_FRAME_LENGTH) {
+            log.warn("68 frame length exceeds maximum allowed: {}", length);
+            buffer.skipBytes(HEADER_LENGTH_68 + 1);
+            return false;
+        }
+
+        if (buffer.readableBytes() < HEADER_LENGTH_68 + 1 + length) {
+            return false; // 数据不足，等待更多数据
+        }
+
+        ByteBuf frame = buffer.readRetainedSlice(HEADER_LENGTH_68 + 1 + length);
+        out.add(new Message68(frame));
+        return true;
+    }
+}

jsowell-netty/src/main/java/com/jsowell/netty/domain/ChargingPileMessage.java

@@ -1,5 +1,8 @@
 package com.jsowell.netty.domain;
 
+import org.apache.commons.lang3.builder.ToStringBuilder;
+import org.apache.commons.lang3.builder.ToStringStyle;
+
 public class ChargingPileMessage {
     private int physicalId;
     private short messageId;
@@ -18,4 +21,15 @@ public class ChargingPileMessage {
     public short getMessageId() { return messageId; }
     public byte getCommand() { return command; }
     public byte[] getData() { return data; }
+
+
+    @Override
+    public String toString() {
+        return new ToStringBuilder(this, ToStringStyle.JSON_STYLE)
+                .append("physicalId", physicalId)
+                .append("messageId", messageId)
+                .append("command", command)
+                .append("data", data)
+                .toString();
+    }
 }

jsowell-netty/src/main/java/com/jsowell/netty/domain/DnyMessage.java

@@ -0,0 +1,15 @@
+package com.jsowell.netty.domain;
+
+import io.netty.buffer.ByteBuf;
+
+public class DnyMessage {
+    private final ByteBuf content;
+
+    public DnyMessage(ByteBuf content) {
+        this.content = content;
+    }
+
+    public ByteBuf getContent() {
+        return content;
+    }
+}

jsowell-netty/src/main/java/com/jsowell/netty/domain/Message68.java

@@ -0,0 +1,15 @@
+package com.jsowell.netty.domain;
+
+import io.netty.buffer.ByteBuf;
+
+public class Message68 {
+    private final ByteBuf content;
+
+    public Message68(ByteBuf content) {
+        this.content = content;
+    }
+
+    public ByteBuf getContent() {
+        return content;
+    }
+}

jsowell-netty/src/main/java/com/jsowell/netty/server/electricbicycles/ChargingPileHandler.java

@@ -3,21 +3,51 @@ package com.jsowell.netty.server.electricbicycles;
 import com.jsowell.netty.domain.ChargingPileMessage;
 import io.netty.channel.ChannelHandler;
 import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.ChannelId;
 import io.netty.channel.ChannelInboundHandlerAdapter;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.stereotype.Component;
 
+import java.net.InetSocketAddress;
 import java.time.Instant;
 import java.nio.ByteBuffer;
 import java.nio.ByteOrder;
+import java.util.concurrent.ConcurrentHashMap;
 
 @ChannelHandler.Sharable
 @Slf4j
 @Component
 public class ChargingPileHandler extends ChannelInboundHandlerAdapter {
 
+    /**
+     * 管理一个全局map，保存连接进服务端的通道数量
+     */
+    private static final ConcurrentHashMap<ChannelId, ChannelHandlerContext> CHANNEL_MAP = new ConcurrentHashMap<>();
+
+    /**
+     * 有客户端连接服务器会触发此函数
+     * 连接被建立并且准备进行通信时被调用
+     */
+    @Override
+    public void channelActive(ChannelHandlerContext ctx) {
+        InetSocketAddress insocket = (InetSocketAddress) ctx.channel().remoteAddress();
+        String clientIp = insocket.getAddress().getHostAddress();
+        int clientPort = insocket.getPort();
+        //获取连接通道唯一标识
+        ChannelId channelId = ctx.channel().id();
+        //如果map中不包含此连接，就保存连接
+        if (CHANNEL_MAP.containsKey(channelId)) {
+            log.info("Handler:{}, 客户端【{}】是连接状态，连接通道数量: {}", this.getClass().getSimpleName(), channelId, CHANNEL_MAP.size());
+        } else {
+            //保存连接
+            CHANNEL_MAP.put(channelId, ctx);
+            log.info("Handler:{}, 客户端【{}】, 连接netty服务器【IP:{}, PORT:{}】, 连接通道数量: {}", this.getClass().getSimpleName(), channelId, clientIp, clientPort, CHANNEL_MAP.size());
+        }
+    }
+
     @Override
     public void channelRead(ChannelHandlerContext ctx, Object msg) {
+        log.info("加载客户端报文=== channelId:{}, mag:{}", ctx.channel().id(), msg.toString());
         if (!(msg instanceof ChargingPileMessage)) {
             return;
         }

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

@@ -3,6 +3,7 @@ package com.jsowell.netty.server.electricbicycles;
 import com.jsowell.netty.decoder.ChargingPileDecoder;
 import com.jsowell.netty.decoder.ProtocolDnyDecoder;
 import com.jsowell.netty.decoder.StartAndLengthFieldFrameDecoder;
+import com.jsowell.netty.decoder.StartAndLengthFieldFrameDecoder2;
 import io.netty.channel.ChannelInitializer;
 import io.netty.channel.ChannelPipeline;
 import io.netty.channel.socket.SocketChannel;
@@ -26,12 +27,12 @@ public class ElectricBicyclesServerChannelInitializer extends ChannelInitializer
 	protected void initChannel(SocketChannel channel) throws Exception {
 		ChannelPipeline pipeline = channel.pipeline();
 		// pipeline.addLast("frameDecoder",new CustomDecoder());
-		pipeline.addLast("frameDecoder", new ChargingPileDecoder());
+		pipeline.addLast("frameDecoder", new StartAndLengthFieldFrameDecoder());
 		pipeline.addLast("decoder", new ByteArrayDecoder());
 		pipeline.addLast("encoder", new ByteArrayDecoder());
 		//读超时时间设置为10s，0表示不监控
 		pipeline.addLast(new IdleStateHandler(60, 0, 0, TimeUnit.SECONDS));
-		pipeline.addLast("handler", electricBicyclesServerHandler);
+		pipeline.addLast("handler", chargingPileHandler);
 	}
 
 }