二进制的协议的测试程序

一、引子

由于要调试二进制私有协议，不想用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)

三、更高层面的抽象：

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

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