游戏测试 python 编写游戏测试机器人客户端 (二)

A_Jian · 2020年12月08日 · 116 次阅读

游戏测试机器人搭建 - Send Actor

Send Actor

Send Actor 初始化

class SendActor(pykka.ThreadingActor):
    '''
    发送消息给服务端
    '''
    def __init__(self, player=None, sock=None):
        super(SendActor, self).__init__()
        self.player = player
        self.socket = sock
        self.wpe = 1
  • 参数传递及调用,在 PlayerActor 里,直接看图 sendActor_args
  • 在 PlayerActor 的 on_receive 里创建 socket/websocket 连接,我的项目是用 websocket,参数传 PlayerActor 的实例和 socket

启动 Send Actor

def on_start(self):
    self.on_heart()
  • 跟 Player Actor 一样,要先调用这里,我在这里启动了心跳方法,可以根据自己的项目类型决定心跳的启动方式或干掉心跳

心跳包

def on_heart(self):
    self.player.sys_count += 1
    if self.player.sys_count >= 590:
        self.actor_ref.tell({MSG_PROTO: {'cmd': 'hall_heart'}})
        self.player.sys_count = 0
    # 心跳这里时间会阻塞100毫秒
    self.actor_ref.tell({MSG_HEART:{'msg':'loop'}})
    time.sleep(0.1)
  • MSG_PROTO:消息类型,发送协议都是这个类型,可自己自定义
  • 这里我是 60 秒跟服务端发一次心跳包
  • {'cmd': 'hall_heart'} : 这是心跳协议,我这里用的 Json 格式的协议,根据自己项目的协议类型更换,如 protobuf,sproto,自定义协议,关于协议转换(序列化和反序列化)这里就不展开讨论,后面再写
  • self.actor_ref.tell({MSG_HEART:{'msg':'loop'}}) : 这一行的代码作用是重复给 SendActor 自己发送消息,在 SendActor 的 on_receive 会接收到消息

序列化和发送数据

def on_receive(self, msg):
    '''
    msg[MSG_PROTO] 打包好的协议数据
    发送包有参数的为元组类型,没有参数则直接发送协议
    '''
    if MSG_PROTO in msg.keys() and msg[MSG_PROTO]:
        proto_id, proto_bin = msg[MSG_PROTO]['cmd'],msg[MSG_PROTO]
        proto_header = {
            'cmd': proto_id,
            'sessionId': self.wpe,
            'ts':int(time.time()*1000),
        }
        proto_header.update(proto_bin)
        buff = json.dumps(proto_header)
        self.socket.send(buff)
    elif msg[MSG_HEART]:
        self.on_heart()
    else:
        print('发过来空数据了')
    if self.wpe is XXX:
        self.wpe = 0
    else:
        self.wpe = self.wpe + 1
  • msg: 这里接收两个数据,协议 ID 和序列化后的协议内容
  • 协议头:proto_header,协议内容和协议头进行合并,再发送给服务端
  • 前面写的心跳方法在这里调用
  • wpe 我这里是指向 sessionID,有些项目需要做序列校验

停止 SendActor

def on_stop(self):
    print('SendActor stop')
  • pykka actor 自带的方法,集中在 PlayerActor 的 on_stop 里调用

又是 log 收集

@GetLog(level='error')
def on_failure(self, exception_type, exception_value, traceback):
    logging.error('SendActor fail => ', exception_type, exception_value, tb.print_tb(traceback))
  • log 模块的使用可以参考"官方说明"或者自行百度查询,案例很多

接下来的是 python 编写游戏测试机器人客户端 (三),跟 SendActor 区别在于反序列化服务端的数据

最后的最后,各位的关注、点赞、收藏、碎银子打赏是对我最大的支持,谢谢大家!
需要源码的小伙伴关注微信公众号 ID:gameTesterGz
或扫描二维码回复机器人脚本即可

如果觉得我的文章对您有用,请随意打赏。您的支持将鼓励我继续创作!
共收到 0 条回复 时间 点赞
A_Jian python 编写游戏测试机器人客户端 (一) 中提及了此贴 12月08日 13:32
需要 登录 后方可回复, 如果你还没有账号请点击这里 注册