Page MenuHomePhabricator
Files F7159223

No OneTemporary
Actions

View Options

diff --git a/sylk/applications/xmppgateway/xmpp/jingle/streams/rtp.py b/sylk/applications/xmppgateway/xmpp/jingle/streams/rtp.py
index 9d117e3..7374829 100644
--- a/sylk/applications/xmppgateway/xmpp/jingle/streams/rtp.py
+++ b/sylk/applications/xmppgateway/xmpp/jingle/streams/rtp.py
@@ -1,441 +1,440 @@
"""
Handling of RTP media streams according to RFC3550, RFC3605, RFC3581,
RFC2833 and RFC3711, RFC3489 and draft-ietf-mmusic-ice-19.
"""
__all__ = ['AudioStream']
from threading import RLock
from application.notification import IObserver, NotificationCenter, NotificationData
from application.python import Null
from zope.interface import implements
from sipsimple.audio import AudioBridge, AudioDevice, IAudioPort
from sipsimple.configuration.settings import SIPSimpleSettings
from sipsimple.core import AudioTransport, PJSIPError, RTPTransport, SIPCoreError
from sipsimple.streams.rtp import RTPStreamEncryption
from sylk.applications.xmppgateway.xmpp.jingle.streams import IMediaStream, InvalidStreamError, MediaStreamRegistrar, UnknownStreamError
class AudioStream(object):
__metaclass__ = MediaStreamRegistrar
implements(IMediaStream, IAudioPort, IObserver)
type = 'audio'
priority = 1
hold_supported = True
def __init__(self):
from sipsimple.application import SIPApplication
self.mixer = SIPApplication.voice_audio_mixer
self.bridge = AudioBridge(self.mixer)
self.device = AudioDevice(self.mixer)
self.notification_center = NotificationCenter()
self.on_hold_by_local = False
self.on_hold_by_remote = False
self.direction = None
self.state = "NULL"
self._transport = None
self._hold_request = None
self._ice_state = "NULL"
self._lock = RLock()
self._rtp_transport = None
self.session = None
self.encryption = RTPStreamEncryption(self)
self._srtp_encryption = None
self._try_ice = False
self._initialized = False
self._done = False
self._failure_reason = None
self.bridge.add(self.device)
# Audio properties
#
@property
def codec(self):
return self._transport.codec if self._transport else None
@property
def consumer_slot(self):
return self._transport.slot if self._transport else None
@property
def producer_slot(self):
return self._transport.slot if self._transport and not self.muted else None
@property
def sample_rate(self):
return self._transport.sample_rate if self._transport else None
@property
def statistics(self):
return self._transport.statistics if self._transport else None
def _get_muted(self):
return self.__dict__.get('muted', False)
def _set_muted(self, value):
if not isinstance(value, bool):
raise ValueError("illegal value for muted property: %r" % (value,))
if value == self.muted:
return
old_producer_slot = self.producer_slot
self.__dict__['muted'] = value
notification_center = NotificationCenter()
data = NotificationData(consumer_slot_changed=False, producer_slot_changed=True, old_producer_slot=old_producer_slot, new_producer_slot=self.producer_slot)
notification_center.post_notification('AudioPortDidChangeSlots', sender=self, data=data)
muted = property(_get_muted, _set_muted)
del _get_muted, _set_muted
# RTP properties
#
@property
def local_rtp_address(self):
return self._rtp_transport.local_rtp_address if self._rtp_transport else None
@property
def local_rtp_port(self):
return self._rtp_transport.local_rtp_port if self._rtp_transport else None
@property
def remote_rtp_address(self):
if self._ice_state == "IN_USE":
return self._rtp_transport.remote_rtp_address_received if self._rtp_transport else None
else:
return self._rtp_transport.remote_rtp_address_sdp if self._rtp_transport else None
@property
def remote_rtp_port(self):
if self._ice_state == "IN_USE":
return self._rtp_transport.remote_rtp_port_received if self._rtp_transport else None
else:
return self._rtp_transport.remote_rtp_port_sdp if self._rtp_transport else None
@property
def local_rtp_candidate_type(self):
return self._rtp_transport.local_rtp_candidate_type if self._rtp_transport else None
@property
def remote_rtp_candidate_type(self):
return self._rtp_transport.remote_rtp_candidate_type if self._rtp_transport else None
@property
def ice_active(self):
return self._ice_state == "IN_USE"
# Generic properties
#
@property
def on_hold(self):
return self.on_hold_by_local or self.on_hold_by_remote
# Public methods
#
@classmethod
def new_from_sdp(cls, session, remote_sdp, stream_index):
# TODO: actually validate the SDP
settings = SIPSimpleSettings()
remote_stream = remote_sdp.media[stream_index]
if remote_stream.media != 'audio':
raise UnknownStreamError
if remote_stream.transport not in ('RTP/AVP', 'RTP/SAVP'):
raise InvalidStreamError("expected RTP/AVP or RTP/SAVP transport in audio stream, got %s" % remote_stream.transport)
local_encryption_policy = 'sdes_optional'
if local_encryption_policy == "sdes_mandatory" and not "crypto" in remote_stream.attributes:
raise InvalidStreamError("SRTP/SDES is locally mandatory but it's not remotely enabled")
if remote_stream.transport == 'RTP/SAVP' and "crypto" in remote_stream.attributes and local_encryption_policy not in ("opportunistic", "sdes_optional", "sdes_mandatory"):
raise InvalidStreamError("SRTP/SDES is remotely mandatory but it's not locally enabled")
supported_codecs = session.account.rtp.audio_codec_list or settings.rtp.audio_codec_list
if not any(codec for codec in remote_stream.codec_list if codec in supported_codecs):
raise InvalidStreamError("no compatible codecs found")
stream = cls()
stream._incoming_remote_sdp = remote_sdp
stream._incoming_stream_index = stream_index
if "zrtp-hash" in remote_stream.attributes:
stream._incoming_stream_encryption = 'zrtp'
elif "crypto" in remote_stream.attributes:
stream._incoming_stream_encryption = 'sdes_mandatory' if remote_stream.transport=='RTP/SAVP' else 'sdes_optional'
else:
stream._incoming_stream_encryption = None
return stream
def initialize(self, session, direction):
with self._lock:
if self.state != "NULL":
raise RuntimeError("AudioStream.initialize() may only be called in the NULL state")
self.state = "INITIALIZING"
self.session = session
local_encryption_policy = 'sdes_optional'
if hasattr(self, "_incoming_remote_sdp"):
# ICE attributes could come at the session level or at the media level
remote_stream = self._incoming_remote_sdp.media[self._incoming_stream_index]
self._try_ice = (remote_stream.has_ice_attributes or self._incoming_remote_sdp.has_ice_attributes) and remote_stream.has_ice_candidates
if self._incoming_stream_encryption is not None and local_encryption_policy == 'opportunistic':
self._srtp_encryption = self._incoming_stream_encryption
else:
self._srtp_encryption = 'zrtp' if local_encryption_policy == 'opportunistic' else local_encryption_policy
del self._incoming_stream_encryption
else:
self._try_ice = True
self._srtp_encryption = 'zrtp' if local_encryption_policy == 'opportunistic' else local_encryption_policy
self._init_rtp_transport()
def get_local_media(self, remote_sdp=None, index=0):
with self._lock:
if self.state not in ["INITIALIZED", "WAIT_ICE", "ESTABLISHED"]:
raise RuntimeError("AudioStream.get_local_media() may only be called in the INITIALIZED, WAIT_ICE or ESTABLISHED states")
if remote_sdp is None:
# offer
old_direction = self._transport.direction
if old_direction is None:
new_direction = "sendrecv"
elif "send" in old_direction:
new_direction = ("sendonly" if (self._hold_request == 'hold' or (self._hold_request is None and self.on_hold_by_local)) else "sendrecv")
else:
new_direction = ("inactive" if (self._hold_request == 'hold' or (self._hold_request is None and self.on_hold_by_local)) else "recvonly")
else:
new_direction = None
return self._transport.get_local_media(remote_sdp, index, new_direction)
def start(self, local_sdp, remote_sdp, stream_index):
with self._lock:
if self.state != "INITIALIZED":
raise RuntimeError("AudioStream.start() may only be called in the INITIALIZED state")
settings = SIPSimpleSettings()
- self._transport.start(local_sdp, remote_sdp, stream_index, no_media_timeout=settings.rtp.timeout,
- media_check_interval=settings.rtp.timeout)
+ self._transport.start(local_sdp, remote_sdp, stream_index, timeout=settings.rtp.timeout)
self._check_hold(self._transport.direction, True)
if self._try_ice:
self.state = 'WAIT_ICE'
else:
self.state = 'ESTABLISHED'
self.notification_center.post_notification('MediaStreamDidStart', sender=self)
def validate_update(self, remote_sdp, stream_index):
with self._lock:
# TODO: implement
return True
def update(self, local_sdp, remote_sdp, stream_index):
with self._lock:
connection = remote_sdp.media[stream_index].connection or remote_sdp.connection
if not self._rtp_transport.ice_active and (connection.address != self._rtp_transport.remote_rtp_address_sdp or self._rtp_transport.remote_rtp_port_sdp != remote_sdp.media[stream_index].port):
settings = SIPSimpleSettings()
old_consumer_slot = self.consumer_slot
old_producer_slot = self.producer_slot
self.notification_center.remove_observer(self, sender=self._transport)
self._transport.stop()
try:
self._transport = AudioTransport(self.mixer, self._rtp_transport, remote_sdp, stream_index, codecs=list(self.session.account.rtp.audio_codec_list or settings.rtp.audio_codec_list))
except SIPCoreError, e:
self.state = "ENDED"
self._failure_reason = e.args[0]
self.notification_center.post_notification('MediaStreamDidFail', sender=self, data=NotificationData(reason=self._failure_reason))
return
self.notification_center.add_observer(self, sender=self._transport)
- self._transport.start(local_sdp, remote_sdp, stream_index, no_media_timeout=settings.rtp.timeout, media_check_interval=settings.rtp.timeout)
+ self._transport.start(local_sdp, remote_sdp, stream_index, timeout=settings.rtp.timeout)
self.notification_center.post_notification('AudioPortDidChangeSlots', sender=self, data=NotificationData(consumer_slot_changed=True, producer_slot_changed=True,
old_consumer_slot=old_consumer_slot, new_consumer_slot=self.consumer_slot,
old_producer_slot=old_producer_slot, new_producer_slot=self.producer_slot))
if connection.address == '0.0.0.0' and remote_sdp.media[stream_index].direction == 'sendrecv':
self._transport.update_direction('recvonly')
self._check_hold(self._transport.direction, False)
self.notification_center.post_notification('RTPStreamDidChangeRTPParameters', sender=self)
else:
new_direction = local_sdp.media[stream_index].direction
self._transport.update_direction(new_direction)
self._check_hold(new_direction, False)
self._hold_request = None
def hold(self):
with self._lock:
if self.on_hold_by_local or self._hold_request == 'hold':
return
if self.state == "ESTABLISHED" and self.direction != "inactive":
self.bridge.remove(self)
self._hold_request = 'hold'
def unhold(self):
with self._lock:
if (not self.on_hold_by_local and self._hold_request != 'hold') or self._hold_request == 'unhold':
return
if self.state == "ESTABLISHED" and self._hold_request == 'hold':
self.bridge.add(self)
self._hold_request = None if self._hold_request == 'hold' else 'unhold'
def deactivate(self):
with self._lock:
self.bridge.stop()
def end(self):
with self._lock:
if not self._initialized or self._done:
return
self._done = True
self.notification_center.post_notification('MediaStreamWillEnd', sender=self)
if self._transport is not None:
self._transport.stop()
self.notification_center.remove_observer(self, sender=self._transport)
self._transport = None
self.notification_center.remove_observer(self, sender=self._rtp_transport)
self._rtp_transport = None
self.state = "ENDED"
self.notification_center.post_notification('MediaStreamDidEnd', sender=self, data=NotificationData(error=self._failure_reason))
self.session = None
def reset(self, stream_index):
with self._lock:
if self.direction == "inactive" and not self.on_hold_by_local:
new_direction = "sendrecv"
self._transport.update_direction(new_direction)
self._check_hold(new_direction, False)
# TODO: do a full reset, re-creating the AudioTransport, so that a new offer
# would contain all codecs and ICE would be renegotiated -Saul
def send_dtmf(self, digit):
with self._lock:
if self.state != "ESTABLISHED":
raise RuntimeError("AudioStream.send_dtmf() cannot be used in %s state" % self.state)
try:
self._transport.send_dtmf(digit)
except PJSIPError, e:
if not e.args[0].endswith("(PJ_ETOOMANY)"):
raise
# Notification handling
#
def handle_notification(self, notification):
handler = getattr(self, '_NH_%s' % notification.name, Null)
handler(notification)
def _NH_RTPTransportDidFail(self, notification):
with self._lock:
self.notification_center.remove_observer(self, sender=notification.sender)
if self.state == "ENDED":
return
self._try_next_rtp_transport(notification.data.reason)
def _NH_RTPTransportDidInitialize(self, notification):
settings = SIPSimpleSettings()
rtp_transport = notification.sender
with self._lock:
if self.state == "ENDED":
return
del self._rtp_args
del self._stun_servers
try:
if hasattr(self, "_incoming_remote_sdp"):
try:
audio_transport = AudioTransport(self.mixer, rtp_transport, self._incoming_remote_sdp, self._incoming_stream_index,
codecs=list(self.session.account.rtp.audio_codec_list or settings.rtp.audio_codec_list))
finally:
del self._incoming_remote_sdp
del self._incoming_stream_index
else:
audio_transport = AudioTransport(self.mixer, rtp_transport, codecs=list(self.session.account.rtp.audio_codec_list or settings.rtp.audio_codec_list))
except SIPCoreError, e:
self.state = "ENDED"
self.notification_center.post_notification('MediaStreamDidNotInitialize', sender=self, data=NotificationData(reason=e.args[0]))
return
self._rtp_transport = rtp_transport
self._transport = audio_transport
self.notification_center.add_observer(self, sender=audio_transport)
self._initialized = True
self.state = "INITIALIZED"
self.notification_center.post_notification('MediaStreamDidInitialize', sender=self)
def _NH_RTPAudioStreamGotDTMF(self, notification):
self.notification_center.post_notification('AudioStreamGotDTMF', sender=self, data=NotificationData(digit=notification.data.digit))
def _NH_RTPAudioTransportDidTimeout(self, notification):
self.notification_center.post_notification('RTPStreamDidTimeout', sender=self)
def _NH_RTPTransportICENegotiationStateDidChange(self, notification):
with self._lock:
if self._ice_state != "NULL" or self.state not in ("INITIALIZING", "INITIALIZED", "WAIT_ICE"):
return
self.notification_center.post_notification('RTPStreamICENegotiationStateDidChange', sender=self, data=notification.data)
def _NH_RTPTransportICENegotiationDidSucceed(self, notification):
with self._lock:
if self.state != "WAIT_ICE":
return
self._ice_state = "IN_USE"
self.state = 'ESTABLISHED'
self.notification_center.post_notification('RTPStreamICENegotiationDidSucceed', sender=self, data=notification.data)
self.notification_center.post_notification('MediaStreamDidStart', sender=self)
def _NH_RTPTransportICENegotiationDidFail(self, notification):
with self._lock:
if self.state != "WAIT_ICE":
return
self._ice_state = "FAILED"
self.state = 'ESTABLISHED'
self.notification_center.post_notification('RTPStreamICENegotiationDidFail', sender=self, data=notification.data)
self.notification_center.post_notification('MediaStreamDidStart', sender=self)
# Private methods
#
def _init_rtp_transport(self, stun_servers=None):
self._rtp_args = dict()
self._rtp_args["encryption"] = self._srtp_encryption
self._rtp_args["use_ice"] = self._try_ice
self._stun_servers = [(None, None)]
if stun_servers:
self._stun_servers.extend(reversed(stun_servers))
self._try_next_rtp_transport()
def _try_next_rtp_transport(self, failure_reason=None):
if self._stun_servers:
stun_address, stun_port = self._stun_servers.pop()
observer_added = False
try:
rtp_transport = RTPTransport(ice_stun_address=stun_address, ice_stun_port=stun_port, **self._rtp_args)
self.notification_center.add_observer(self, sender=rtp_transport)
observer_added = True
rtp_transport.set_INIT()
except SIPCoreError, e:
if observer_added:
self.notification_center.remove_observer(self, sender=rtp_transport)
self._try_next_rtp_transport(e.args[0])
else:
self.state = "ENDED"
self.notification_center.post_notification('MediaStreamDidNotInitialize', sender=self, data=NotificationData(reason=failure_reason))
def _check_hold(self, direction, is_initial):
was_on_hold_by_local = self.on_hold_by_local
was_on_hold_by_remote = self.on_hold_by_remote
was_inactive = self.direction == "inactive"
self.direction = direction
inactive = self.direction == "inactive"
self.on_hold_by_local = was_on_hold_by_local if inactive else direction == "sendonly"
self.on_hold_by_remote = "send" not in direction
if (is_initial or was_on_hold_by_local or was_inactive) and not inactive and not self.on_hold_by_local and self._hold_request != 'hold':
self.bridge.add(self)
if not was_on_hold_by_local and self.on_hold_by_local:
self.notification_center.post_notification('RTPStreamDidChangeHoldState', sender=self, data=NotificationData(originator="local", on_hold=True))
if was_on_hold_by_local and not self.on_hold_by_local:
self.notification_center.post_notification('RTPStreamDidChangeHoldState', sender=self, data=NotificationData(originator="local", on_hold=False))
if not was_on_hold_by_remote and self.on_hold_by_remote:
self.notification_center.post_notification('RTPStreamDidChangeHoldState', sender=self, data=NotificationData(originator="remote", on_hold=True))
if was_on_hold_by_remote and not self.on_hold_by_remote:
self.notification_center.post_notification('RTPStreamDidChangeHoldState', sender=self, data=NotificationData(originator="remote", on_hold=False))

File Metadata

Mime Type
text/x-diff
Expires
Sat, Nov 23, 3:27 AM (13 h, 2 m)
Storage Engine
blob
Storage Format
Raw Data
Storage Handle
3408698
Default Alt Text
(21 KB)

Event Timeline