使用aiortc通过TCP从server向client推流时遇RTCIceTransport错误

问题根源分析 #

出现InvalidStateError('RTCIceTransport is closed')主要是因为：

• 信令流程不完整，仅处理单次信令消息，未处理所有ICE候选交换
• 连接建立后立即关闭PeerConnection，未等待媒体流传输
• 媒体轨道的时间戳处理缺失，导致流同步异常
• 客户端轨道接收逻辑未正确启动，且UI渲染存在参数错误

具体修复步骤 #

1. 完善信令处理逻辑 #

服务器和客户端都需要循环处理信令消息，直到信令通道关闭，确保所有ICE候选都能被正确添加。

2. 延迟PeerConnection关闭时机 #

服务器需等待ICE连接建立完成，且媒体流传输结束后再关闭；客户端需等待媒体流结束或ICE连接断开后再关闭。

3. 修复媒体轨道的时间戳处理 #

自定义VideoStreamTrack的recv方法中必须生成正确的pts和time_base，否则aiortc无法正确处理流同步。

4. 修复客户端的视频渲染逻辑 #

cv2.imshow需传入转换后的ndarray图像，而非原始VideoFrame；同时要在on_track中启动轨道的接收循环。

修正后的代码 #

Server.py #

import asyncio
import cv2
from aiortc import MediaStreamTrack, RTCPeerConnection, RTCSessionDescription, RTCIceCandidate
from aiortc.contrib.media import MediaRelay
from aiortc.contrib.signaling import TcpSocketSignaling
from av import VideoFrame


class VideoStreamTrack(MediaStreamTrack):
    kind = "video"

    def __init__(self, video_path):
        super().__init__()
        self.video_path = video_path
        self.cap = cv2.VideoCapture(video_path)
        self.fps = self.cap.get(cv2.CAP_PROP_FPS) or 30

    async def recv(self):
        ret, img = self.cap.read()
        if not ret:
            self.cap.release()
            raise ConnectionError("Video stream ended")
        
        # 生成正确的时间戳，确保流同步
        pts, time_base = await self.next_timestamp()
        frame = VideoFrame.from_ndarray(img, format="bgr24")
        frame.pts = pts
        frame.time_base = time_base
        
        # 按视频帧率控制发送间隔
        await asyncio.sleep(1 / self.fps)
        return frame


async def handle_signaling(pc, signaling):
    while True:
        try:
            message = await signaling.receive()
            if isinstance(message, RTCSessionDescription):
                await pc.setRemoteDescription(message)
            elif isinstance(message, RTCIceCandidate):
                await pc.addIceCandidate(message)
            elif message is None:
                break
        except Exception as e:
            print(f"信令处理错误: {e}")
            break


async def serve_video(pc, signaling):
    relay = MediaRelay()
    video_path = "test.mp4"
    video_track = VideoStreamTrack(video_path)
    pc.addTrack(relay.subscribe(video_track))

    # 创建并发送offer
    offer = await pc.createOffer()
    await pc.setLocalDescription(offer)
    await signaling.send(pc.localDescription)

    # 启动信令处理任务
    signaling_task = asyncio.create_task(handle_signaling(pc, signaling))

    # 等待ICE连接建立
    while pc.iceConnectionState not in ["connected", "completed", "failed"]:
        await asyncio.sleep(0.1)
    
    if pc.iceConnectionState == "failed":
        print("ICE连接失败")
        await pc.close()
        return

    print("ICE连接已建立，开始推流")

    try:
        # 持续推送直到视频结束
        while True:
            await video_track.recv()
    except ConnectionError:
        print("视频流已结束")
    finally:
        # 等待信令任务完成
        signaling_task.cancel()
        await pc.close()


async def main():
    pc = RTCPeerConnection()

    # 注册ICE状态变化回调
    @pc.on("iceconnectionstatechange")
    def on_iceconnectionstatechange():
        print(f"ICE连接状态: {pc.iceConnectionState}")

    signaling = TcpSocketSignaling("127.0.0.1", 8080)
    await signaling.connect()

    try:
        await serve_video(pc, signaling)
    finally:
        await signaling.close()


if __name__ == "__main__":
    asyncio.run(main())

Client.py #

import asyncio
import cv2
from aiortc import MediaStreamTrack, RTCPeerConnection, RTCSessionDescription, RTCIceCandidate
from aiortc.contrib.signaling import TcpSocketSignaling


class VideoStreamTrack(MediaStreamTrack):
    kind = "video"

    def __init__(self, track):
        super().__init__()
        self.track = track

    async def recv(self):
        frame = await self.track.recv()
        img = frame.to_ndarray(format="bgr24")
        cv2.imshow("Received Video", img)
        # 必须调用waitKey，否则窗口无法刷新
        if cv2.waitKey(1) & 0xFF == ord('q'):
            raise ConnectionError("用户终止播放")
        return frame


async def handle_signaling(pc, signaling):
    while True:
        try:
            message = await signaling.receive()
            if isinstance(message, RTCSessionDescription):
                await pc.setRemoteDescription(message)
                # 如果是offer则创建并发送answer
                if message.type == "offer":
                    answer = await pc.createAnswer()
                    await pc.setLocalDescription(answer)
                    await signaling.send(pc.localDescription)
            elif isinstance(message, RTCIceCandidate):
                await pc.addIceCandidate(message)
            elif message is None:
                break
        except Exception as e:
            print(f"信令处理错误: {e}")
            break


async def receive_video(pc, signaling):
    # 存储视频轨道引用
    video_track = None

    @pc.on("track")
    def on_track(track):
        nonlocal video_track
        if track.kind == "video":
            video_track = VideoStreamTrack(track)
            print("已收到视频轨道")

    @pc.on("iceconnectionstatechange")
    async def on_iceconnectionstatechange():
        print(f"ICE连接状态: {pc.iceConnectionState}")
        if pc.iceConnectionState == "failed":
            print("ICE连接失败")
            await pc.close()

    # 启动信令处理任务
    signaling_task = asyncio.create_task(handle_signaling(pc, signaling))

    # 等待视频轨道初始化
    while video_track is None:
        await asyncio.sleep(0.1)

    try:
        # 持续接收视频帧
        while True:
            await video_track.recv()
    except (ConnectionError, asyncio.CancelledError):
        print("播放结束")
    finally:
        cv2.destroyAllWindows()
        signaling_task.cancel()
        await pc.close()


async def main():
    pc = RTCPeerConnection()
    signaling = TcpSocketSignaling("127.0.0.1", 8080)
    await signaling.connect()

    try:
        await receive_video(pc, signaling)
    finally:
        await signaling.close()


if __name__ == "__main__":
    asyncio.run(main())

额外注意事项 #

• 确保test.mp4文件存在于代码同级目录，或修改路径为绝对路径
• 运行时先启动服务器，再启动客户端
• 客户端窗口可按q键终止播放

内容的提问来源于stack exchange，提问作者10may