mediaproxy 代码理解解读

media relay中的几个概念：

SessionManager 用于管理所有session

Session 一次通话是一个Session。一个Session包含一个或者多个MediaStream

MediaStream 是指一个独立的音频流、视频流或者其他数据流。一个MediaStream包含俩MediaParty，俩MediaSubStream

MediaParty 是指参与方，主叫方或者被叫方

MediaSubStream 是指rtp流或者rtcp流

MediaSubParty 是指一个rtp流或者rtcp流下面的主叫方和被叫方

MediaParty 对象建立时，建立udp侦听端口，侦听rtp或者rtcp端口的来包。先过来的一般是stun包，用于探测ice通路。在中继被决定使用后才会有rtp或者rtcp数据

	

		


	

	

		            try:
	

	

		                self.listener_rtp = reactor.listenUDP(port_rtp, StreamListenerProtocol(), interface=Config.relay_ip)
	

	

		                self.listener_rtcp = reactor.listenUDP(port_rtcp, StreamListenerProtocol(), interface=Config.relay_ip)
	

	

		
	

	

		


	

MediaParty先于MediaSubStream创建，MediaSubStream创建时会传入MediaParty创建的listener。

	

		


	

	

		        self.rtp = MediaSubStream(self, self.caller.listener_rtp, self.callee.listener_rtp)
	

	

		        self.rtcp = MediaSubStream(self, self.caller.listener_rtcp, self.callee.listener_rtcp)
	

	

		
	

	

		


	

而MediaSubParty是由MediaSubStream创建的，创建时会传入listener。

	

		


	

	

		        self.caller = MediaSubParty(self, listener_caller)
	

	

		        self.callee = MediaSubParty(self, listener_callee)
	

	

		
	

	

		


	

listener的call back函数是指向MediaSubParty类的got_data

	

		


	

	

		        self.listener.protocol.cb_func = self.got_data
	

	

		
	

	

		


	

got_data，会用于判断是否第一次接收到与是否stun包。

	

		


	

	

		        if (host, port) == tuple(self.remote):
	

	

		            if self.remote.obsolete:
	

	

		                # the received packet matches the previously used IP/port,
	

	

		                # which has been made obsolete, so ignore it
	

	

		                return
	

	

		        else:
	

	

		            if self.remote.in_use:
	

	

		                # the received packet is different than the recorded IP/port,
	

	

		                # so we will discard it
	

	

		                return
	

	

		            # we have learnt the remote IP/port
	

	

		            self.remote.host, self.remote.port = host, port
	

	

		            self.remote.in_use = True
	

	

		            log.debug("Got traffic information for stream: %s" % self.substream.stream)
	

	

		        is_stun, is_binding_request = _stun_test(data)
	

	

		        self.substream.send_data(self, data, is_stun)
	

	

		
	

	

		


	

而MediaSubParty创建时启动了一个定时器，用于侦听是否一定时间内（默认90）没有rtp数据。

	

		


	

	

		        if self.timer and self.timer.active():
	

	

		            self.timer.cancel()
	

	

		        self.timer = reactor.callLater(Config.stream_timeout, self.substream.expired, "no-traffic timeout", Config.stream_timeout)
	

	

		
	

	

		


	

如果收到第一个rtp数据，该定时器就会取消。同时调用substream.check_create_conntrack，检查是否改成conntrack来转发（内核态转发，效率更高）

	

		


	

	

		        if not self.remote.got_rtp and not is_stun:
	

	

		            # This is the first RTP packet received
	

	

		            self.remote.got_rtp = True
	

	

		            if self.timer:
	

	

		                if self.timer.active():
	

	

		                    self.timer.cancel()
	

	

		                self.timer = None
	

	

		
	

	

		            self.substream.check_create_conntrack()
	

	

		
	

	

		


	

conntrack启动的条件，是收发双方都已经有rtp数据发送过来

	

		


	

	

		        if self.caller.remote.in_use and self.caller.remote.got_rtp and self.callee.remote.in_use and self.callee.remote.got_rtp:
	

	

		            self.forwarding_rule = _conntrack.ForwardingRule(self.caller.remote, self.caller.local, self.callee.remote, self.callee.local, self.stream.session.mark)
	

	

		            self.forwarding_rule.expired_func = self.conntrack_expired
	

	

		
	

	

		


	

如果收或者发其中一方没有rtp数据，会回调MediaSubStream的expired，再回调stream.substream_expired，导致整个session的中继结束

	

		


	

	

		    def expired(self, reason, timeout_wait):
	

	

		        self._stop_relaying()
	

	

		        self.stream.substream_expired(self, reason, timeout_wait)
	

	

		
	

	

		


	

需要思考的地方：

一、假如客户端用的是ICE aggressive，有可能出现一边走中继，一边走p2p

二、一方可能一直没有rtp数据发出，例如一方没有摄像头