二进制的协议的测试程序

2024-05-25 22:52

本文主要是介绍二进制的协议的测试程序,希望对大家解决编程问题提供一定的参考价值,需要的开发者们随着小编来一起学习吧!

一、引子

由于要调试二进制私有协议,不想用C++重头到尾写,用C++写工程量有点大,因此想找一个比较简单的工具,postman无法实现,外界的几乎找不到合适的工具,只能考虑手写一个。
前面写了一个python通过tcp协议发送二进制数据,那个是第一个版本,虽然也能做事,只是每次都要来一遍,太麻烦。
手写最快的觉得还是python,主要是熟。
不过真的开写时发现也是各种坑,要是调试一两个也就罢了,要调试很多,每个都是重头写,也能写吐。二、抽象出对请求类,接收类的处理考虑一种方案,包头,公共包头这些通信最好一次就好了,各种包的通信其实也可以封装抽象出来。实际上各包之间唯一的区别就是各个协议的发送类和接收类不一样,
这点无法共用,其它关于打包成二进制数据,由二进制数据解析成类对象,借助python强大的功能,可以直接用buffer(),unpack()全部搞定,当然,各类的成员
变量是多大是少不了得自己先定义下来了

理出来后最重要的是需要把类结构首先要转成标识这个类实际有多大的字符串集,

@staticmethod
def generate_buffer_format(fields):format_str = '<'for field_name, field_type in fields:if issubclass(field_type, ctypes.Array) and issubclass(field_type._type_, ctypes.c_char):array_length = field_type._length_format_str += f'{array_length}s'else:format_str += field_type._type_return format_str

如上代码实现了各种类最终会用类似'<BBI'等字符串表示出来,然后通过

struct.pack将其解析成二进制的buf

转成buf和由buf转换赋值给类实例中各成员变量,代码如下:

    def buffer(self) -> bytes:"""生成一个表达类实例中各成员变量大小的格式串,然后获取类实例对应每个成员的值列表,最后有了格式串,有了成员值列表,将之组装成二进制buf"""format_str = self.generate_buffer_format(self._fields_)values = self.get_fields_values(self._fields_)return struct.pack(format_str, *values)
    def unpack(self, bin_data):"""生成一个表达类实例中各成员变量大小的格式串,然后根据这样的格式串将二进制buf生成一组数值data,然后将这一组数值按顺序赋给类实例中各成员变量"""format_str = self.generate_buffer_format(self._fields_)data = struct.unpack(format_str, bin_data)for field, value in zip(self._fields_, data):setattr(self, field[0], value)

转码过成完后就是通信了,通信最主要的是不要管转码的过程,只负责数据发送和接收,然后调用各种类,由它们完成转码的过程:

如下所示:


def communication(command_id, ReqClass: Type[StructWithBuffer], RespClass: Type[StructWithBuffer]):"""二进制私有协议的通信过程,传入协议号,该协议号对应的请求类和响应类,完成请求和响应的通信过程"""client_head = ClientHead()client_pub_head = ClientPubHead()client_pub_head.iType = command_idreq_obj = ReqClass()client_head.iLen = len(client_pub_head.buffer()) + len(req_obj.buffer())final_data = client_head.buffer() + client_pub_head.buffer() + req_obj.buffer()# 建立 TCP 连接并发送数据with (socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s):s.connect((TCP_IP, TCP_PORT))s.sendall(final_data)recv_data = s.recv(ctypes.sizeof(ClientHead))recv_head = ClientHead()recv_head.unpack(recv_data)# 输出解析后的数据logging.info("recv_iLen:{}".format(recv_head.iLen))# 解析第二个包的数据并写入到 ClientPubHead 类的成员变量中recv_data = s.recv(recv_head.iLen)recv_pub_head = ClientPubHead()recv_pub_head.unpack(recv_data[:ctypes.sizeof(ClientPubHead)])logging.info("recv_iType:{}".format(recv_pub_head.iType))body_len = recv_head.iLen - ctypes.sizeof(ClientPubHead)if body_len <= 8:body_data = recv_data[ctypes.sizeof(ClientPubHead):]logging.info("recv_data_len:{},recv_data:{}".format(body_len, body_data))else:body_data = recv_data[ctypes.sizeof(ClientPubHead):]count = body_len / ctypes.sizeof(RespClass)logging.info(f"recv_body_len:{body_len}, body_count:{count}")for i in range(int(count)):resp_size = ctypes.sizeof(RespClass)data = body_data[i*resp_size: (i+1)*resp_size]info = RespClass()info.unpack(data)logging.info(f"{info}")logging.info("communication end.")

如此,就实现了一个很单纯的发送,接收,然后调用类名将之实例化,由各自的类实例完成转码。

测试代码相当简单:

if __name__ == '__main__':from binary_req_base import communicationimport logginglogging.basicConfig(format='%(asctime)s:%(module)s[%(lineno)d] %(message)s',level=logging.DEBUG)communication(3011, ReqActiveStock, RespActiveStock)

三、更高层面的抽象:

这个是一个半成品,即,只需定义各个类,即可直接完成收发的过程了,更强大的是写一个工具,然后只需在配置文件中定义各类的变量的大小即可。

另外,如果接收类是一个复杂的,外层先是一层,然后内存再有若干层,明显这个框架就不合适。需要对通信函数做重载,把这复杂的包含关系抽象出来。

这篇关于二进制的协议的测试程序的文章就介绍到这儿,希望我们推荐的文章对编程师们有所帮助!



http://www.chinasem.cn/article/1002855

相关文章

【Linux】应用层http协议

一、HTTP协议 1.1 简要介绍一下HTTP        我们在网络的应用层中可以自己定义协议,但是,已经有大佬定义了一些现成的,非常好用的应用层协议,供我们直接使用,HTTP(超文本传输协议)就是其中之一。        在互联网世界中,HTTP(超文本传输协议)是一个至关重要的协议,他定义了客户端(如浏览器)与服务器之间如何进行通信,以交换或者传输超文本(比如HTML文档)。

【Go】go连接clickhouse使用TCP协议

离开你是傻是对是错 是看破是软弱 这结果是爱是恨或者是什么 如果是种解脱 怎么会还有眷恋在我心窝 那么爱你为什么                      🎵 黄品源/莫文蔚《那么爱你为什么》 package mainimport ("context""fmt""log""time""github.com/ClickHouse/clickhouse-go/v2")func main(

2024.9.8 TCP/IP协议学习笔记

1.所谓的层就是数据交换的深度,电脑点对点就是单层,物理层,加上集线器还是物理层,加上交换机就变成链路层了,有地址表,路由器就到了第三层网络层,每个端口都有一个mac地址 2.A 给 C 发数据包,怎么知道是否要通过路由器转发呢?答案:子网 3.将源 IP 与目的 IP 分别同这个子网掩码进行与运算****,相等则是在一个子网,不相等就是在不同子网 4.A 如何知道,哪个设备是路由器?答案:在 A

Modbus-RTU协议

一、协议概述 Modbus-RTU(Remote Terminal Unit)是一种基于主从架构的通信协议,采用二进制数据表示,消息中的每个8位字节含有两个4位十六进制字符。它主要通过RS-485、RS-232、RS-422等物理接口实现数据的传输,传输距离远、抗干扰能力强、通信效率高。 二、报文结构 一个标准的Modbus-RTU报文通常包含以下部分: 地址域:单个字节,表示从站设备

通信工程学习:什么是2ASK/BASK二进制振幅键控

2ASK/BASK:二进制振幅键控         2ASK/BASK二进制振幅键控是一种数字调制技术,其全称是二进制振幅键控(Binary Amplitude Shift Keying)。该技术通过改变载波的振幅来传递二进制数字信息,而载波的频率和相位则保持不变。以下是关于2ASK/BASK二进制振幅键控的详细解释: 一、2ASK/BASK二进制振幅键控的基本原理 1、振幅键控:

网络原理之TCP协议(万字详解!!!)

目录 前言 TCP协议段格式 TCP协议相关特性 1.确认应答 2.超时重传 3.连接管理(三次握手、四次挥手) 三次握手(建立TCP连接) 四次挥手(断开连接)  4.滑动窗口 5.流量控制 6.拥塞控制 7.延迟应答 8.捎带应答  9.基于字节流 10.异常情况的处理 小结  前言 在前面,我们已经讲解了有关UDP协议的相关知识,但是在传输层,还有

1 模拟——67. 二进制求和

1 模拟 67. 二进制求和 给你两个二进制字符串 a 和 b ,以二进制字符串的形式返回它们的和。 示例 1:输入:a = "11", b = "1"输出:"100"示例 2:输入:a = "1010", b = "1011"输出:"10101" 算法设计 可以从低位到高位(从后向前)计算,用一个变量carry记录进位,如果有字符没处理完或者有进位,则循环处理。两个字符串对

DNS协议基础笔记

1.定义 DNS(Domain Name System,域名系统)是互联网的一项核心服务,它作为将域名和 IP 地址相互映射的一个分布式数据库,能够使人更方便地访问互联网。 2.域名解析过程 当用户在浏览器中输入一个域名,浏览器首先会检查自己的缓存中是否有该域名对应的 IP 地址。本地 DNS 服务器收到查询请求后,首先会检查自己的缓存中是否有该域名对应的 IP 地址。根域名服务器收到查询请

4G模块、WIFI模块、NBIOT模块通过AT指令连接华为云物联网服务器(MQTT协议)

MQTT协议概述 MQTT(Message Queuing Telemetry Transport)是一种轻量级的消息传输协议,它被设计用来提供一对多的消息分发和应用之间的通讯,尤其适用于远程位置的设备和高延迟或低带宽的网络。MQTT协议基于客户端-服务器架构,客户端可以订阅任意数量的主题,并可以发布消息到这些主题。服务器(通常称为MQTT Broker)则负责接受来自客户端的连接请求,并转发消

Leetcode67---二进制求和

https://leetcode.cn/problems/add-binary/description/ 给出的两个二进制,我们可以从最后开始往前运算。 给当前短的一位前面补充0即可。 class Solution {public String addBinary(String a, String b) {//给的就是二进制字符串 最后一位开始遍历 如果没有就补充0?StringBuil