Implementing a simple TCP custom protocol based on Kotlin

1. Development Background

If you want to be an excellent Android developer, you need a complete knowledge system. Here, let us grow together to become what we want to be~.

Our project requires the development of a smart hardware. It sends instructions from the web backend to a desktop application, and then the desktop program controls different hardware devices to implement business operations. From the web backend to the desktop, a WebSocket long link is used for maintenance, and from the desktop program to each hardware device, a TCP long link is also used for maintenance.

[[342699]]

This article actually describes the communication between desktop programs and various hardware.

2. Custom communication protocol

First, a general TCP network protocol needs to be designed.

The network protocol structure is as follows

 + --------------+---------------+----------------+---------------+-----------+----------+  
 | magic number (4) | version (1) | serialization method (1) | command (1) | data length (4) | data (n) |
     + --------------+---------------+----------------+---------------+-----------+----------+  

• Magic number: 4 bytes. This project uses 20200803 (the day it was written). To prevent the port from being accidentally called, we take the first 4 bytes after receiving the message and compare them with the magic number. If they are different, we directly reject and close the connection.
• Version number: 1 byte, only indicates the version number of the protocol, which is convenient for use when the protocol is upgraded
• Serialization method: 1 byte, indicating how to convert a Java object into binary data and how to deserialize it.
• Command: 1 byte, indicating the intention of the message (such as taking a photo, taking a video, heartbeat, App upgrade, etc.). A maximum of 2^8 commands are supported.
• Data length: 4 bytes, indicating the length of the data after this field. Supports up to 2^32 bits.
• Data: The content of specific data.

According to the network protocol designed above, define an abstract class Packet:

 abstract class Packet {
 var magic: Int ? = MAGIC_NUMBER // magic number
 var version:Byte = 1 // Version number, the current protocol version number is 1
 abstract val serializeMethod:Byte // serialization method
 abstract val command:Byte // Command for Watcher to communicate with App
 }

As many Packets as there are instructions, you need to define as many Packets. Let's take the heartbeat Packet as an example and define a HeartBeatPacket:

 data class HeartBeatPacket(var msg:String = "ping" ,
                           override val serializeMethod: Byte = Serialize.JSON,
                           override val command: Byte = Commands.HEART_BEAT) : Packet() {
 }

HeartBeatPacket is initiated by the TCP client, received by the TCP server and returned to the client.

Each Packet class contains the serialization method used by the Packet.

 /**
 * List of constants for serialization
 */
 interface Serialize {
 companion object {
        const val JSON: Byte = 0
 }}

Each Packet also contains its corresponding command. The following is the Commands instruction set, which supports 256 instructions.

 /**
 * Instruction set, supports 256 instructions from -128 to 127
 */
 interface Commands {
 companion object {
 /**
 * Heartbeat Packet
 */
        const val HEART_BEAT: Byte = 0
 /**
 * Login (App needs to tell Watcher: cameraPosition location)
 */
        const val LOGIN: Byte = 1
 ...... }}

Since a custom protocol is used, the message must be encoded and decoded, and PacketManager is responsible for these tasks.

Encoding is the process of assembling messages according to the protocol structure. Similarly, decoding is the reverse process.

 /**
 * Message management class, encode and decode messages
 */
 object PacketManager {
    fun encode(packet: Packet):ByteBuf = encode(ByteBufAllocator. DEFAULT , packet)
    fun encode(alloc:ByteBufAllocator, packet: Packet) = encode(alloc.ioBuffer(), packet)
    fun encode(buf: ByteBuf, packet: Packet): ByteBuf {
        val serializer = SerializerFactory.getSerializer(packet.serializeMethod)
        val bytes: ByteArray = serializer.serialize(packet)
 //Assemble message: magic number (4 bytes) + version number (1 byte) + serialization method (1 byte) + instruction (1 byte) + data length (4 bytes) + data (N bytes)
        buf.writeInt(MAGIC_NUMBER)
        buf.writeByte(packet.version.toInt())
        buf.writeByte(packet.serializeMethod.toInt())
        buf.writeByte(packet.command.toInt())
        buf.writeInt( bytes.size )
 buf.writeBytes(bytes)
 return buf
 }
    fun decode(buf:ByteBuf): Packet {
 buf.skipBytes(4) // The magic number is checked by a separate Handler
 buf.skipBytes(1)
        val serializationMethod = buf.readByte()
        val serializer = SerializerFactory.getSerializer(serializationMethod)
        val command = buf.readByte()
        valclazz = PacketFactory.getPacket(command)
 val length = buf.readInt() // length of data
 val bytes = ByteArray(length) // Define the character array to be read
 buf.readBytes(bytes)
 return serializer.deserialize(clazz, bytes)
 }
 }

3. TCP Server

How to start TCP service

 fun execute () {
    boss = NioEventLoopGroup() worker = NioEventLoopGroup() val bootstrap = ServerBootstrap()
    bootstrap.group (boss, worker) .channel (NioServerSocketChannel::class.java)
            . option (ChannelOption.SO_BACKLOG, 100)
            .childOption(ChannelOption.SO_KEEPALIVE, true )
            .childOption(ChannelOption.SO_REUSEADDR, true )
            .childOption(ChannelOption.TCP_NODELAY, true )
            .childHandler(object : ChannelInitializer<NioSocketChannel>() {
                @Throws(Exception::class)
                override fun initChannel(nioSocketChannel: NioSocketChannel) {
                    val pipeline = nioSocketChannel.pipeline()
                    pipeline.addLast(ServerIdleHandler()) pipeline.addLast(MagicNumValidator()) pipeline.addLast(PacketCodecHandler) pipeline.addLast(HeartBeatHandler) pipeline.addLast(ResponseHandler) } }) val future: ChannelFuture = bootstrap.bind(TCP_PORT)
    future.addListener(object : ChannelFutureListener {
        @Throws(Exception::class)
        override fun operationComplete(channelFuture: ChannelFuture) {
            if (channelFuture.isSuccess) {
                logInfo(logger, "TCP Server is starting..." )
 } else {
                logError(logger,channelFuture.cause(), "TCP Server failed" )
            } } }) }

Among them, ServerIdleHandler: means that if no heartbeat is received within 5 minutes, the connection will be disconnected.

 class ServerIdleHandler : IdleStateHandler(0, 0, HERT_BEAT_TIME) {
    private val logger: Logger = LoggerFactory.getLogger(ServerIdleHandler::class.java)
 @Throws(Exception::class)
    override fun channelIdle(ctx: ChannelHandlerContext, evt: IdleStateEvent) {
 logInfo(logger) { ctx.channel(). close () "If no heartbeat is received within $HERT_BEAT_TIME seconds, the connection will be disconnected"  
        } } companion object {
        private const val HERT_BEAT_TIME = 300
 }}

MagicNumValidator: Magic number validation for TCP packets.

 class MagicNumValidator : LengthFieldBasedFrameDecoder( Int .MAX_VALUE, LENGTH_FIELD_OFFSET, LENGTH_FIELD_LENGTH) {
    private val logger: Logger = LoggerFactory.getLogger(this.javaClass)
 @Throws(Exception::class)
    override fun decode(ctx: ChannelHandlerContext, ` in `: ByteBuf): Any ? {
 if (` in `.getInt(` in `.readerIndex()) !== MAGIC_NUMBER) { // If the magic number check fails, close the connection
 logInfo(logger, "Magic number check failed" )
            ctx.channel(). close ()
 return   null  
 }
 return super.decode(ctx, ` in `)
 }
 companion object {
        private const val LENGTH_FIELD_OFFSET = 7
        private const val LENGTH_FIELD_LENGTH = 4
 }
 }

PacketCodecHandler: Handler that parses the message.

PacketCodecHandler inherits from ByteToMessageCodec. It is used to process byte-to-message and message-to-byte, so as to decode byte messages into POJO or encode POJO messages into bytes.

 @ChannelHandler.Sharable
 object PacketCodecHandler : MessageToMessageCodec<ByteBuf, Packet>() { override fun encode(ctx: ChannelHandlerContext, msg: Packet, list: MutableList< Any >) {
        val byteBuf = ctx.channel().alloc().ioBuffer()
        PacketManager.encode(byteBuf, msg) list. add (byteBuf) } override fun decode(ctx: ChannelHandlerContext, msg: ByteBuf, list: MutableList< Any >) {
        list.add (PacketManager.decode(msg)); }}

HeartBeatHandler: Heartbeat Handler, receives "ping" from TCP client, and returns "pong" to the client.

 @ChannelHandler.Sharable
 object HeartBeatHandler : SimpleChannelInboundHandler<HeartBeatPacket>(){ private val logger: Logger = LoggerFactory.getLogger(this.javaClass)
    override fun channelRead0(ctx: ChannelHandlerContext, msg: HeartBeatPacket) {
 logInfo(logger, "Received heartbeat packet: ${GsonUtils.toJson(msg)}" )
 msg.msg = "pong" // Return pong to the client
        ctx.writeAndFlush(msg)
 }
 }

ResponseHandler: A general-purpose Handler that receives instructions from TCP clients. It can query the corresponding Handler based on the corresponding instructions and process its commands.

 object ResponseHandler: SimpleChannelInboundHandler<Packet>() {
    private val logger: Logger = LoggerFactory.getLogger(this.javaClass)
    private val handlerMap: ConcurrentHashMap<Byte, SimpleChannelInboundHandler< out Packet>> = ConcurrentHashMap()
 init {
        handlerMap[LOGIN] = LoginHandler ...... handlerMap[ERROR] = ErrorHandler } override fun channelRead0(ctx: ChannelHandlerContext, msg: Packet) {
 logInfo(logger, "Received command from client: ${msg.command}" )
        val handler: SimpleChannelInboundHandler< out Packet>? = handlerMap[msg.command]
 handler?.let { logInfo(logger, "Found the Handler that responded to the instruction: ${it.javaClass.simpleName}" )
 it.channelRead(ctx, msg) } ?: logInfo(logger, "Handler that responds to command not found" )
    } @Throws(Exception::class)
    override fun channelInactive(ctx: ChannelHandlerContext) {
        val insocket = ctx.channel().remoteAddress() as InetSocketAddress
        val clientIP = insocket.address.hostAddress
        val clientPort = insocket.port
 logError(logger, "Client disconnected: $clientIP : $clientPort" )
        super.channelInactive(ctx)
 }}

4. TCP Client

Simulating a client implementation

 val topLevelClass = object : Any () {}.javaClass.enclosingClass
 val logger: Logger = LoggerFactory.getLogger(topLevelClass)fun main() {
    val worker = NioEventLoopGroup()
    val bootstrap = Bootstrap()
    bootstrap.group (worker).channel(NioSocketChannel::class.java)
            .handler(object : ChannelInitializer<SocketChannel>() {
                @Throws(Exception::class)
                override fun initChannel(channel: SocketChannel) {
                    channel.pipeline().addLast(PacketCodecHandler) channel.pipeline().addLast(ClientIdleHandler()) channel.pipeline().addLast(ClientLogin()) } }) val future: ChannelFuture = bootstrap.connect ( "127.0.0.1" , TCP_PORT).addListener(object : ChannelFutureListener {
        @Throws(Exception::class)
        override fun operationComplete(channelFuture: ChannelFuture) {
            if (channelFuture.isSuccess()) {
                logInfo(logger, "connect to server success!" )
 } else {
                logger.info( "failed to connect the server! " )
                System.exit(0)
            } } }) try {
 future.channel().closeFuture().sync() logInfo(logger, "Disconnected from the server!" )
    } catch (e: InterruptedException) {
        e.printStackTrace() }}

Among them, PacketCodecHandler is the same as the Handler used by the server to parse the message.

ClientIdleHandler: The client implements heartbeat and sends a heartbeat every 30 seconds.

 class ClientIdleHandler : IdleStateHandler(0, 0, HEART_BEAT_TIME) {
    private val logger = LoggerFactory.getLogger(ClientIdleHandler::class.java)
 @Throws(Exception::class)
    override fun channelIdle(ctx: ChannelHandlerContext, evt: IdleStateEvent?) {
 logInfo(logger, "Sending heartbeat...." )
        ctx.writeAndFlush(HeartBeatPacket()) } companion object {
        private const val HEART_BEAT_TIME = 30
 }}

ClientLogin: Handler for logging into the server.

 @ChannelHandler.Sharable
 class ClientLogin: ChannelInboundHandlerAdapter() { private val logger: Logger = LoggerFactory.getLogger(this.javaClass)
 @Throws(Exception::class)
    override fun channelActive(ctx: ChannelHandlerContext) {
        val packet: LoginPacket = LoginPacket()
        logInfo(logger, "packet = ${GsonUtils.toJson(packet)}" )
        val byteBuf = PacketManager.encode(packet)
        ctx.channel().writeAndFlush(byteBuf) }}

V. Conclusion

This time, the logic of the desktop program I developed is not complicated. It only needs to receive instructions from the Web background and then interact with various devices.

After receiving the command from the Web side, the command is sent to each device through TCP through Guava's EventBus, and needs to be converted into the corresponding Packet when sending. Therefore, the core module is this TCP custom protocol.