Server-py/sri-server-bg03/lib/ConnectionTest2004.py

"""
echo "执行：进入调试" \
&& project_path="/home/sri/repositories/repositories/sri-project.demo-py/sri-dino-pyserver01" \
&& cd ${project_path} \
&& sudo docker exec -it sri-dino-pyserver01 bash

python3 /home/sri/repositories/repositories/sri-project.demo-py/sri-dino-pyserver01/lib/ConnectionTest.py
"""
import socket
import struct
import traceback
import time

import sys
import importlib

# --- for linux
sys.path.append('/home/sri/repositories/repositories/sri-project.demo-py/sri-dino-pyserver01')
sys.path.append('/home/sri/repositories/repositories/sri-project.demo-py/3rdparty')
# host, port = '127.0.0.1', 20916
host, port = '127.0.0.1', 20917

# --- for windows
# sys.path.append(r'E:\casper\repositories\repositories\sri-project.demo-py\sri-server-bg03')
# sys.path.append(r'E:\casper\repositories\repositories\sri-project.demo-py\3rdparty')
# host, port = '58.34.94.178', 20916
# host, port = '58.34.94.178', 20916
# host, port = '58.34.94.178', 20917

# --- import
protobuf = importlib.import_module(f"xprotobuf.protocol_pb2")
methods = importlib.import_module(f"xlib")


def test():
    head_sequence = '<hh'  # 字节序规则
    head_size = struct.calcsize(head_sequence)

    try:
        client = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        # client.bind(('localhost', port))
        client.connect((host, port))

        # --- send 2004
        object = protobuf.Offer()
        object.index = 1
        object.peer = 1000004
        object.type = 'offer'
        object.sdp = 'aabbcc'
        head = 2004, object.ByteSize()
        head_bytestream = struct.pack(head_sequence, *head)
        body_bytestream = object.SerializeToString()
        send_bytestream = head_bytestream + body_bytestream
        print(f"Sent data: {send_bytestream}")
        client.sendall(send_bytestream)

        # 等待接收
        while True:
            pass


    except Exception as exception:

        print(f"#exception: {exception.__class__.__name__}")
        print(f"#traceback: {traceback.format_exc()}")

    finally:
        client.close()


if __name__ == "__main__":
    test()