diff --git a/sylk/applications/webrtcgateway/handler.py b/sylk/applications/webrtcgateway/handler.py index c37b5b3..8451b53 100644 --- a/sylk/applications/webrtcgateway/handler.py +++ b/sylk/applications/webrtcgateway/handler.py @@ -1,1660 +1,1662 @@ import hashlib import json import random import os import time from application.notification import IObserver, NotificationCenter, NotificationData from application.python import Null, limit from application.python.weakref import defaultweakobjectmap from application.system import makedirs, unlink from collections import deque from eventlib import coros, proc from itertools import count from sipsimple.configuration.settings import SIPSimpleSettings from sipsimple.core import SIPURI, FromHeader, ToHeader, Credentials from sipsimple.lookup import DNSLookup, DNSLookupError from sipsimple.streams import MediaStreamRegistry from sipsimple.threading import run_in_thread, run_in_twisted_thread from sipsimple.threading.green import call_in_green_thread, run_in_green_thread from shutil import copyfileobj, rmtree from string import maketrans from twisted.internet import reactor from typing import Generic, Container, Iterable, Sized, TypeVar, Dict, Set, Optional, Union from werkzeug.exceptions import InternalServerError from zope.interface import implements from sylk.accounts import DefaultAccount from sylk.session import Session from . import push from .configuration import GeneralConfig, get_room_config, ExternalAuthConfig from .janus import JanusBackend, JanusError, JanusSession, SIPPluginHandle, VideoroomPluginHandle from .logger import ConnectionLogger, VideoroomLogger from .models import sylkrtc, janus from .storage import TokenStorage from .auth import AuthHandler class AccountInfo(object): # noinspection PyShadowingBuiltins def __init__(self, id, password, display_name=None, user_agent=None): self.id = id self.password = password self.display_name = display_name self.user_agent = user_agent self.registration_state = None self.janus_handle = None # type: Optional[SIPPluginHandle] self.contact_params = {} self.auth_handle = None self.auth_state = None @property def uri(self): return 'sip:' + self.id @property def user_data(self): return dict(username=self.uri, display_name=self.display_name, user_agent=self.user_agent, ha1_secret=self.password, contact_params=self.contact_params) class SessionPartyIdentity(object): def __init__(self, uri, display_name=None): self.uri = uri self.display_name = display_name # todo: might need to replace this auto-resetting descriptor with a timer in case we need to know when the slow link state expired class SlowLinkState(object): def __init__(self): self.slow_link = False self.last_reported = 0 class SlowLinkDescriptor(object): __timeout__ = 30 # 30 seconds def __init__(self): self.values = defaultweakobjectmap(SlowLinkState) def __get__(self, instance, owner): if instance is None: return self state = self.values[instance] if state.slow_link and time.time() - state.last_reported > self.__timeout__: state.slow_link = False return state.slow_link def __set__(self, instance, value): state = self.values[instance] if value: state.last_reported = time.time() state.slow_link = bool(value) def __delete__(self, instance): raise AttributeError('Attribute cannot be deleted') class SIPSessionInfo(object): slow_download = SlowLinkDescriptor() slow_upload = SlowLinkDescriptor() # noinspection PyShadowingBuiltins def __init__(self, id): self.id = id self.direction = None self.state = None self.account = None # type: Optional[AccountInfo] self.local_identity = None # type: Optional[SessionPartyIdentity] self.remote_identity = None # type: Optional[SessionPartyIdentity] self.janus_handle = None # type: Optional[SIPPluginHandle] self.slow_download = False self.slow_upload = False def init_outgoing(self, account, destination): self.account = account self.direction = 'outgoing' self.state = 'connecting' self.local_identity = SessionPartyIdentity(account.id) self.remote_identity = SessionPartyIdentity(destination) def init_incoming(self, account, originator, originator_display_name=''): self.account = account self.direction = 'incoming' self.state = 'connecting' self.local_identity = SessionPartyIdentity(account.id) self.remote_identity = SessionPartyIdentity(originator, originator_display_name) class VideoroomSessionInfo(object): slow_download = SlowLinkDescriptor() slow_upload = SlowLinkDescriptor() # noinspection PyShadowingBuiltins def __init__(self, id, owner, janus_handle): self.type = None # publisher / subscriber self.id = id self.owner = owner # type: ConnectionHandler self.janus_handle = janus_handle # type: VideoroomPluginHandle self.chat_handler = None # type: Optional[VideoroomChatHandler] self.account = None # type: Optional[AccountInfo] self.room = None # type: Optional[Videoroom] self.bitrate = None self.parent_session = None # type: Optional[VideoroomSessionInfo] # for subscribers this is their main session (the one used to join), for publishers is None self.publisher_id = None # janus publisher ID for publishers / publisher session ID for subscribers self.slow_download = False self.slow_upload = False self.feeds = PublisherFeedContainer() # keeps references to all the other participant's publisher feeds that we subscribed to def init_publisher(self, account, room): self.type = 'publisher' self.account = account self.room = room self.bitrate = room.config.max_bitrate self.chat_handler = VideoroomChatHandler(session=self) def init_subscriber(self, publisher_session, parent_session): assert publisher_session.type == parent_session.type == 'publisher' self.type = 'subscriber' self.publisher_id = publisher_session.id self.parent_session = parent_session self.account = parent_session.account self.room = parent_session.room self.bitrate = self.room.config.max_bitrate def __repr__(self): return '<{0.__class__.__name__}: type={0.type!r} id={0.id!r} janus_handle={0.janus_handle!r}>'.format(self) class PublisherFeedContainer(object): """A container for the other participant's publisher sessions that we have subscribed to""" def __init__(self): self._publishers = set() self._id_map = {} # map publisher.id -> publisher and publisher.publisher_id -> publisher def add(self, session): assert session not in self._publishers assert session.id not in self._id_map and session.publisher_id not in self._id_map self._publishers.add(session) self._id_map[session.id] = self._id_map[session.publisher_id] = session def discard(self, item): # item can be any of session, session.id or session.publisher_id session = self._id_map[item] if item in self._id_map else item if item in self._publishers else None if session is not None: self._publishers.discard(session) self._id_map.pop(session.id, None) self._id_map.pop(session.publisher_id, None) def remove(self, item): # item can be any of session, session.id or session.publisher_id session = self._id_map[item] if item in self._id_map else item self._publishers.remove(session) self._id_map.pop(session.id) self._id_map.pop(session.publisher_id) def pop(self, item): # item can be any of session, session.id or session.publisher_id session = self._id_map[item] if item in self._id_map else item self._publishers.remove(session) self._id_map.pop(session.id) self._id_map.pop(session.publisher_id) return session def clear(self): self._publishers.clear() self._id_map.clear() def __len__(self): return len(self._publishers) def __iter__(self): return iter(self._publishers) def __getitem__(self, key): return self._id_map[key] def __contains__(self, item): return item in self._id_map or item in self._publishers class Videoroom(object): - def __init__(self, uri): + def __init__(self, uri, audio, video): self.id = random.getrandbits(32) # janus needs numeric room names self.uri = uri + self.audio = audio + self.video = video self.config = get_room_config(uri) self.log = VideoroomLogger(self) self._active_participants = [] self._sessions = set() # type: Set[VideoroomSessionInfo] self._id_map = {} # type: Dict[Union[str, int], VideoroomSessionInfo] # map session.id -> session and session.publisher_id -> session self._shared_files = [] self._raised_hands = [] if self.config.record: makedirs(self.config.recording_dir, 0o755) self.log.info('created (recording on)') else: self.log.info('created') @property def active_participants(self): return self._active_participants @active_participants.setter def active_participants(self, participant_list): unknown_participants = set(participant_list).difference(self._id_map) if unknown_participants: raise ValueError('unknown participant session id: {}'.format(', '.join(unknown_participants))) if self._active_participants != participant_list: self._active_participants = participant_list self.log.info('active participants: {}'.format(', '.join(self._active_participants) or None)) self._update_bitrate() @property def raised_hands(self): return self._raised_hands @raised_hands.setter def raised_hands(self, session_id): if session_id in self._raised_hands: self.log.info('{session} lowers hand '.format(session=session_id)) self._raised_hands.remove(session_id) else: self.log.info('{session} raises hand '.format(session=session_id)) self._raised_hands.append(session_id) def add(self, session): assert session not in self._sessions assert session.publisher_id is not None assert session.publisher_id not in self._id_map and session.id not in self._id_map self._sessions.add(session) self._id_map[session.id] = self._id_map[session.publisher_id] = session self.log.info('{session.account.id} has joined the room'.format(session=session)) self._update_bitrate() if self._active_participants: session.owner.send(sylkrtc.VideoroomConfigureEvent(session=session.id, active_participants=self._active_participants, originator='videoroom')) if self._shared_files: session.owner.send(sylkrtc.VideoroomFileSharingEvent(session=session.id, files=self._shared_files)) if self._raised_hands: session.owner.send(sylkrtc.VideoroomRaisedHandsEvent(session=session.id, raised_hands=self._raised_hands)) # noinspection DuplicatedCode def discard(self, session): if session in self._sessions: self._sessions.discard(session) self._id_map.pop(session.id, None) self._id_map.pop(session.publisher_id, None) self.log.info('{session.account.id} has left the room'.format(session=session)) if session.id in self._active_participants: self._active_participants.remove(session.id) self.log.info('active participants: {}'.format(', '.join(self._active_participants) or None)) for session in self._sessions: session.owner.send(sylkrtc.VideoroomConfigureEvent(session=session.id, active_participants=self._active_participants, originator='videoroom')) self._update_bitrate() # noinspection DuplicatedCode def remove(self, session): self._sessions.remove(session) self._id_map.pop(session.id) self._id_map.pop(session.publisher_id) self.log.info('{session.account.id} has left the room'.format(session=session)) if session.id in self._active_participants: self._active_participants.remove(session.id) self.log.info('active participants: {}'.format(', '.join(self._active_participants) or None)) for session in self._sessions: session.owner.send(sylkrtc.VideoroomConfigureEvent(session=session.id, active_participants=self._active_participants, originator='videoroom')) self._update_bitrate() def clear(self): for session in self._sessions: self.log.info('{session.account.id} has left the room'.format(session=session)) self._active_participants = [] self._shared_files = [] self._sessions.clear() self._id_map.clear() def allow_uri(self, uri): config = self.config if config.access_policy == 'allow,deny': return config.allow.match(uri) and not config.deny.match(uri) else: return not config.deny.match(uri) or config.allow.match(uri) def add_file(self, upload_request): self._write_file(upload_request) def get_file(self, filename): path = os.path.join(self.config.filesharing_dir, filename) if os.path.exists(path): return path else: raise LookupError('file does not exist') @staticmethod def _fix_path(path): name, extension = os.path.splitext(path) for x in count(0, step=-1): path = '{}{}{}'.format(name, x or '', extension) if not os.path.exists(path) and not os.path.islink(path): return path @run_in_thread('file-io') def _write_file(self, upload_request): makedirs(self.config.filesharing_dir) path = self._fix_path(os.path.join(self.config.filesharing_dir, upload_request.shared_file.filename)) upload_request.shared_file.filename = os.path.basename(path) try: with open(path, 'wb') as output_file: copyfileobj(upload_request.content, output_file) except (OSError, IOError): upload_request.had_error = True unlink(path) self._write_file_done(upload_request) @run_in_twisted_thread def _write_file_done(self, upload_request): if upload_request.had_error: upload_request.deferred.errback(InternalServerError('could not save file')) else: self._shared_files.append(upload_request.shared_file) for session in self._sessions: session.owner.send(sylkrtc.VideoroomFileSharingEvent(session=session.id, files=[upload_request.shared_file])) upload_request.deferred.callback('OK') def cleanup(self): self._remove_files() @run_in_thread('file-io') def _remove_files(self): rmtree(self.config.filesharing_dir, ignore_errors=True) def _update_bitrate(self): if self._sessions: if self._active_participants: # todo: should we use max_bitrate / 2 or max_bitrate for each active participant if there are 2 active participants? active_participant_bitrate = self.config.max_bitrate // len(self._active_participants) other_participant_bitrate = 100000 self.log.debug('participant bitrate is {} (active) / {} (others)'.format(active_participant_bitrate, other_participant_bitrate)) for session in self._sessions: if session.id in self._active_participants: bitrate = active_participant_bitrate else: bitrate = other_participant_bitrate if session.bitrate != bitrate: session.bitrate = bitrate session.janus_handle.message(janus.VideoroomUpdatePublisher(bitrate=bitrate), async=True) else: bitrate = self.config.max_bitrate // limit(len(self._sessions) - 1, min=1) self.log.debug('participant bitrate is {}'.format(bitrate)) for session in self._sessions: if session.bitrate != bitrate: session.bitrate = bitrate session.janus_handle.message(janus.VideoroomUpdatePublisher(bitrate=bitrate), async=True) # todo: make Videoroom be a context manager that is retained/released on enter/exit and implement __nonzero__ to be different from __len__ # todo: so that a videoroom is not accidentally released by the last participant leaving while a new participant waits to join # todo: this needs a new model for communication with janus and the client that is pseudo-synchronous (uses green threads) def __len__(self): return len(self._sessions) def __iter__(self): return iter(self._sessions) def __getitem__(self, key): return self._id_map[key] def __contains__(self, item): return item in self._id_map or item in self._sessions SessionT = TypeVar('SessionT', SIPSessionInfo, VideoroomSessionInfo) class SessionContainer(Sized, Iterable[SessionT], Container[SessionT], Generic[SessionT]): def __init__(self): self._sessions = set() self._id_map = {} # map session.id -> session and session.janus_handle.id -> session def add(self, session): assert session not in self._sessions assert session.id not in self._id_map and session.janus_handle.id not in self._id_map self._sessions.add(session) self._id_map[session.id] = self._id_map[session.janus_handle.id] = session def discard(self, item): # item can be any of session, session.id or session.janus_handle.id session = self._id_map[item] if item in self._id_map else item if item in self._sessions else None if session is not None: self._sessions.discard(session) self._id_map.pop(session.id, None) self._id_map.pop(session.janus_handle.id, None) def remove(self, item): # item can be any of session, session.id or session.janus_handle.id session = self._id_map[item] if item in self._id_map else item self._sessions.remove(session) self._id_map.pop(session.id) self._id_map.pop(session.janus_handle.id) def pop(self, item): # item can be any of session, session.id or session.janus_handle.id session = self._id_map[item] if item in self._id_map else item self._sessions.remove(session) self._id_map.pop(session.id) self._id_map.pop(session.janus_handle.id) return session def clear(self): self._sessions.clear() self._id_map.clear() def __len__(self): return len(self._sessions) def __iter__(self): return iter(self._sessions) def __getitem__(self, key): return self._id_map[key] def __contains__(self, item): return item in self._id_map or item in self._sessions class OperationName(str): __normalizer__ = maketrans('-', '_') @property def normalized(self): return self.translate(self.__normalizer__) class Operation(object): __slots__ = 'type', 'name', 'data' __types__ = 'request', 'event' # noinspection PyShadowingBuiltins def __init__(self, type, name, data): if type not in self.__types__: raise ValueError("Can't instantiate class {.__class__.__name__} with unknown type: {!r}".format(self, type)) self.type = type self.name = OperationName(name) self.data = data class APIError(Exception): pass class GreenEvent(object): def __init__(self): self._event = coros.event() def set(self): if self._event.ready(): return self._event.send(True) def is_set(self): return self._event.ready() def clear(self): if self._event.ready(): self._event.reset() def wait(self): return self._event.wait() # noinspection PyPep8Naming class ConnectionHandler(object): implements(IObserver) janus = JanusBackend() def __init__(self, protocol): self.protocol = protocol self.device_id = hashlib.md5(protocol.peer).digest().encode('base64').rstrip('=\n') self.janus_session = None # type: Optional[JanusSession] self.accounts_map = {} # account ID -> account self.account_handles_map = {} # Janus handle ID -> account self.sip_sessions = SessionContainer() # type: SessionContainer[SIPSessionInfo] # incoming and outgoing SIP sessions self.videoroom_sessions = SessionContainer() # type: SessionContainer[VideoroomSessionInfo] # publisher and subscriber sessions in video rooms self.ready_event = GreenEvent() self.resolver = DNSLookup() self.proc = proc.spawn(self._operations_handler) self.operations_queue = coros.queue() self.log = ConnectionLogger(self) self.state = None self._stop_pending = False @run_in_green_thread def start(self): self.state = 'starting' try: self.janus_session = JanusSession() except Exception as e: self.state = 'failed' self.log.warning('could not create session, disconnecting: %s' % e) if self._stop_pending: # if stop was already called it means we were already disconnected self.stop() else: self.protocol.disconnect(3000, unicode(e)) else: self.state = 'started' self.ready_event.set() if self._stop_pending: self.stop() else: self.send(sylkrtc.ReadyEvent()) def stop(self): if self.state in (None, 'starting'): self._stop_pending = True return self.state = 'stopping' self._stop_pending = False if self.proc is not None: # Kill the operation's handler proc first, in order to not have any operations active while we cleanup. self.proc.kill() # Also proc.kill() will switch to another green thread, which is another reason to do it first so that self.proc = None # we do not switch to another green thread in the middle of the cleanup with a partially deleted handler if self.ready_event.is_set(): # Do not explicitly detach the janus plugin handles before destroying the janus session. Janus runs each request in a different # thread, so making detach and destroy request without waiting for the detach to finish can result in errors from race conditions. # Because we do not want to wait for them, we will rely instead on the fact that janus automatically detaches the plugin handles # when it destroys a session, so we only remove our event handlers and issue a destroy request for the session. for account_info in self.accounts_map.values(): if account_info.janus_handle is not None: self.janus.set_event_handler(account_info.janus_handle.id, None) for session in self.sip_sessions: if session.janus_handle is not None: self.janus.set_event_handler(session.janus_handle.id, None) for session in self.videoroom_sessions: if session.janus_handle is not None: self.janus.set_event_handler(session.janus_handle.id, None) if session.chat_handler is not None: notification_center = NotificationCenter() notification_center.remove_observer(self, sender=session.chat_handler) session.chat_handler.end() session.chat_handler = None if session in session.room: # We need to check if the room can be destroyed, else this will never happen reactor.callLater(2, call_in_green_thread, self._maybe_destroy_videoroom_after_disconnect, session.room) session.room.discard(session) session.feeds.clear() self.janus_session.destroy() # this automatically detaches all plugin handles associated with it, no need to manually do it # cleanup self.ready_event.clear() self.accounts_map.clear() self.account_handles_map.clear() self.sip_sessions.clear() self.videoroom_sessions.clear() self.janus_session = None self.protocol = None self.state = 'stopped' def handle_notification(self, notification): handler = getattr(self, '_NH_%s' % notification.name, Null) handler(notification) def handle_message(self, message): try: request = sylkrtc.SylkRTCRequest.from_message(message) except sylkrtc.ProtocolError as e: self.log.error(str(e)) except Exception as e: self.log.error('{request_type}: {exception!s}'.format(request_type=message['sylkrtc'], exception=e)) if 'transaction' in message: self.send(sylkrtc.ErrorResponse(transaction=message['transaction'], error=str(e))) else: operation = Operation(type='request', name=request.sylkrtc, data=request) self.operations_queue.send(operation) def send(self, message): if self.protocol is not None: self.protocol.sendMessage(json.dumps(message.__data__)) # internal methods (not overriding / implementing the protocol API) def _cleanup_session(self, session): # should only be called from a green thread. if self.janus_session is None: # The connection was closed, there is noting to do return if session in self.sip_sessions: self.sip_sessions.remove(session) if session.direction == 'outgoing': # Destroy plugin handle for outgoing sessions. For incoming ones it's the same as the account handle, so don't session.janus_handle.detach() def _cleanup_videoroom_session(self, session): # should only be called from a green thread. if self.janus_session is None: # The connection was closed, there is noting to do return if session in self.videoroom_sessions: self.videoroom_sessions.remove(session) if session.type == 'publisher': notification_center = NotificationCenter() notification_center.remove_observer(self, sender=session.chat_handler) session.room.discard(session) session.feeds.clear() session.janus_handle.detach() session.chat_handler.end() self._maybe_destroy_videoroom(session.room) else: session.parent_session.feeds.discard(session.publisher_id) session.janus_handle.detach() def _maybe_destroy_videoroom(self, videoroom): # should only be called from a green thread. if self.protocol is None or self.janus_session is None: # The connection was closed, there is nothing to do return if videoroom in self.protocol.factory.videorooms and not videoroom: self.protocol.factory.videorooms.remove(videoroom) videoroom.cleanup() with VideoroomPluginHandle(self.janus_session, event_handler=self._handle_janus_videoroom_event) as videoroom_handle: videoroom_handle.destroy(room=videoroom.id) videoroom.log.info('destroyed') def _maybe_destroy_videoroom_after_disconnect(self, videoroom): # should only be called from a green thread. if self.protocol is None and not videoroom: videoroom.cleanup() videoroom.log.info('destroyed') def _lookup_sip_proxy(self, uri): # The proxy dance: Sofia-SIP seems to do a DNS lookup per SIP message when a domain is passed # as the proxy, so do the resolution ourselves and give it pre-resolver proxy URL. Since we use # caching to avoid long delays, we randomize the results matching the highest priority route's # transport. proxy = GeneralConfig.outbound_sip_proxy if proxy is not None: sip_uri = SIPURI(host=proxy.host, port=proxy.port, parameters={'transport': proxy.transport}) else: sip_uri = SIPURI.parse('sip:%s' % uri) settings = SIPSimpleSettings() try: routes = self.resolver.lookup_sip_proxy(sip_uri, settings.sip.transport_list).wait() except DNSLookupError as e: raise DNSLookupError('DNS lookup error: {exception!s}'.format(exception=e)) if not routes: raise DNSLookupError('DNS lookup error: no results found') route = random.choice([r for r in routes if r.transport == routes[0].transport]) self.log.debug('DNS lookup for SIP proxy for {} yielded {}'.format(uri, route)) # Build a proxy URI Sofia-SIP likes return 'sips:{route.address}:{route.port}'.format(route=route) if route.transport == 'tls' else str(route.uri) def _callid_to_uuid(self, callid): hexa = hashlib.md5(callid.encode()).hexdigest() uuidv4 = '%s-%s-%s-%s-%s' % (hexa[:8], hexa[8:12], hexa[12:16], hexa[16:20], hexa[20:]) return uuidv4 def _handle_janus_sip_event(self, event): operation = Operation(type='event', name='janus-sip', data=event) self.operations_queue.send(operation) def _handle_janus_videoroom_event(self, event): operation = Operation(type='event', name='janus-videoroom', data=event) self.operations_queue.send(operation) def _operations_handler(self): self.ready_event.wait() while True: operation = self.operations_queue.wait() handler = getattr(self, '_OH_' + operation.type) handler(operation) del operation, handler def _OH_request(self, operation): handler = getattr(self, '_RH_' + operation.name.normalized) request = operation.data try: handler(request) except (APIError, DNSLookupError, JanusError) as e: self.log.error('{operation.name}: {exception!s}'.format(operation=operation, exception=e)) self.send(sylkrtc.ErrorResponse(transaction=request.transaction, error=str(e))) except Exception as e: self.log.exception('{operation.type} {operation.name}: {exception!s}'.format(operation=operation, exception=e)) self.send(sylkrtc.ErrorResponse(transaction=request.transaction, error='Internal error')) else: self.send(sylkrtc.AckResponse(transaction=request.transaction)) def _OH_event(self, operation): handler = getattr(self, '_EH_' + operation.name.normalized) try: handler(operation.data) except Exception as e: self.log.exception('{operation.type} {operation.name}: {exception!s}'.format(operation=operation, exception=e)) # Request handlers def _RH_ping(self, request): pass def _RH_account_add(self, request): if request.account in self.accounts_map: raise APIError('Account {request.account} already added'.format(request=request, peer=self.protocol.peer)) # check if domain is acceptable domain = request.account.partition('@')[2] if not {'*', domain}.intersection(GeneralConfig.sip_domains): raise APIError('SIP domain not allowed: %s' % domain) # Create and store our mapping account_info = AccountInfo(request.account, request.password, request.display_name, request.user_agent) # get the auth config for domain account_info.auth_handle = AuthHandler(account_info, self) self.accounts_map[account_info.id] = account_info self.log.info('{peer} added account {request.account} using {request.user_agent}'.format(request=request, peer=self.protocol.peer)) def _RH_account_remove(self, request): try: account_info = self.accounts_map.pop(request.account) except KeyError: raise APIError('Unknown account specified for remove: {request.account}'.format(request=request, peer=self.protocol.peer)) # cleanup in case the client didn't unregister before removing the account if account_info.janus_handle is not None: account_info.janus_handle.detach() self.account_handles_map.pop(account_info.janus_handle.id) self.log.info('{peer} removed account {request.account}'.format(request=request, peer=self.protocol.peer)) def _RH_account_register(self, request): try: account_info = self.accounts_map[request.account] except KeyError: raise APIError('Unknown account specified for register: {request.account}'.format(request=request, peer=self.protocol.peer)) proxy = self._lookup_sip_proxy(request.account) if account_info.janus_handle is not None: # Destroy the existing plugin handle account_info.janus_handle.detach() self.account_handles_map.pop(account_info.janus_handle.id) account_info.janus_handle = None # Create a plugin handle account_info.janus_handle = SIPPluginHandle(self.janus_session, event_handler=self._handle_janus_sip_event) self.account_handles_map[account_info.janus_handle.id] = account_info if ExternalAuthConfig.enable: account_info.auth_handle.authenticate(proxy) else: account_info.janus_handle.register(account_info, proxy=proxy) self.log.info('{peer} registering account {request.account}...'.format(request=request, peer=self.protocol.peer)) def _RH_account_unregister(self, request): try: account_info = self.accounts_map[request.account] except KeyError: raise APIError('Unknown account specified for unregister: {request.account}'.format(request=request, peer=self.protocol.peer)) if account_info.janus_handle is not None: account_info.janus_handle.detach() self.account_handles_map.pop(account_info.janus_handle.id) account_info.janus_handle = None if 'pn_tok' in account_info.contact_params: storage = TokenStorage() storage.remove(request.account, account_info.contact_params['pn_tok']) self.log.info('{peer} unregistered account {request.account}'.format(request=request, peer=self.protocol.peer)) def _RH_account_devicetoken(self, request): if request.account not in self.accounts_map: raise APIError('Unknown account specified for token: {request.account}'.format(request=request, peer=self.protocol.peer)) if request.token is not None: account_info = self.accounts_map[request.account] account_info.contact_params = { 'pn_app': request.app, 'pn_tok': request.token, 'pn_type': request.platform, 'pn_device': request.device, 'pn_silent': str(int(request.silent is True)) # janus expects a string } storage = TokenStorage() storage.add(request.account, account_info.contact_params, account_info.user_agent) self.log.info('{peer} added token for account {request.account} on {request.platform} device {request.device})'.format(request=request, peer=self.protocol.peer)) def _RH_session_create(self, request): if request.session in self.sip_sessions: raise APIError('Session ID {request.session} already in use'.format(request=request, peer=self.protocol.peer)) try: account_info = self.accounts_map[request.account] except KeyError: raise APIError('Unknown account specified: {request.account}'.format(request=request, peer=self.protocol.peer)) proxy = self._lookup_sip_proxy(request.uri) # Create a new plugin handle and 'register' it, without actually doing so janus_handle = SIPPluginHandle(self.janus_session, event_handler=self._handle_janus_sip_event) try: janus_handle.call(account_info, uri=request.uri, sdp=request.sdp, proxy=proxy) except Exception: janus_handle.detach() raise session_info = SIPSessionInfo(request.session) session_info.janus_handle = janus_handle session_info.init_outgoing(account_info, request.uri) self.sip_sessions.add(session_info) self.log.info('{peer} outgoing session {request.session} from {request.account} to {request.uri}'.format(request=request, peer=self.protocol.peer)) def _RH_session_answer(self, request): try: session_info = self.sip_sessions[request.session] except KeyError: raise APIError('Unknown session {request.session}'.format(request=request, peer=self.protocol.peer)) if session_info.direction != 'incoming': raise APIError('Cannot answer outgoing session {request.session}'.format(request=request, peer=self.protocol.peer)) if session_info.state != 'connecting': raise APIError('Invalid state for answering session {session.id}: {session.state}'.format(session=session_info, peer=self.protocol.peer)) session_info.janus_handle.accept(sdp=request.sdp) self.log.info('{peer} incoming session {session.id} answered'.format(session=session_info, peer=self.protocol.peer)) def _RH_session_trickle(self, request): try: session_info = self.sip_sessions[request.session] except KeyError: raise APIError('Unknown session {request.session}'.format(request=request, peer=self.protocol.peer)) if session_info.state == 'terminated': raise APIError('Session {request.session} is terminated'.format(request=request, peer=self.protocol.peer)) session_info.janus_handle.trickle(request.candidates) if not request.candidates: self.log.debug('session {session.id} negotiated ICE'.format(session=session_info, peer=self.protocol.peer)) def _RH_session_terminate(self, request): try: session_info = self.sip_sessions[request.session] except KeyError: raise APIError('Unknown session {request.session}'.format(request=request, peer=self.protocol.peer)) if session_info.state not in ('connecting', 'progress', 'early_media', 'accepted', 'established'): raise APIError('Invalid state for terminating session {session.id}: {session.state}'.format(session=session_info, peer=self.protocol.peer)) if session_info.direction == 'incoming' and session_info.state == 'connecting': session_info.janus_handle.decline() else: session_info.janus_handle.hangup() self.log.info('{peer} {session.direction} session {session.id} will terminate'.format(session=session_info, peer=self.protocol.peer)) def _RH_videoroom_join(self, request): if request.session in self.videoroom_sessions: raise APIError('Session ID {request.session} already in use'.format(request=request, peer=self.protocol.peer)) try: account_info = self.accounts_map[request.account] except KeyError: raise APIError('Unknown account specified: {request.account}'.format(request=request, peer=self.protocol.peer)) try: videoroom = self.protocol.factory.videorooms[request.uri] except KeyError: - videoroom = Videoroom(request.uri) + videoroom = Videoroom(request.uri, request.audio, request.video) self.protocol.factory.videorooms.add(videoroom) if not videoroom.allow_uri(request.account): self._maybe_destroy_videoroom(videoroom) raise APIError('{request.account} is not allowed to join room {request.uri}'.format(request=request, peer=self.protocol.peer)) try: videoroom_handle = VideoroomPluginHandle(self.janus_session, event_handler=self._handle_janus_videoroom_event) try: try: videoroom_handle.create(room=videoroom.id, config=videoroom.config, publishers=10) except JanusError as e: if e.code != 427: # 427 means room already exists raise - videoroom_handle.join(room=videoroom.id, sdp=request.sdp, display_name=account_info.display_name) + videoroom_handle.join(room=videoroom.id, sdp=request.sdp, display_name=account_info.display_name, audio=videoroom.audio, video=videoroom.video) except Exception: videoroom_handle.detach() raise except Exception: self._maybe_destroy_videoroom(videoroom) raise videoroom_session = VideoroomSessionInfo(request.session, owner=self, janus_handle=videoroom_handle) videoroom_session.init_publisher(account=account_info, room=videoroom) self.videoroom_sessions.add(videoroom_session) notification_center = NotificationCenter() notification_center.add_observer(self, sender=videoroom_session.chat_handler) videoroom_session.chat_handler.start() self.send(sylkrtc.VideoroomSessionProgressEvent(session=videoroom_session.id)) self.log.info('{peer} session {request.session} from account {request.account} to video room {request.uri} created'.format(request=request, peer=self.protocol.peer)) def _RH_videoroom_leave(self, request): try: videoroom_session = self.videoroom_sessions[request.session] except KeyError: raise APIError('Unknown video room session: {request.session}'.format(request=request, peer=self.protocol.peer)) videoroom_session.janus_handle.leave() self.send(sylkrtc.VideoroomSessionTerminatedEvent(session=videoroom_session.id)) # safety net in case we do not get any answer for the leave request # todo: to be adjusted later after pseudo-synchronous communication with janus is implemented reactor.callLater(2, call_in_green_thread, self._cleanup_videoroom_session, videoroom_session) self.log.info('{peer} leaving video room session {request.session}'.format(request=request, peer=self.protocol.peer)) def _RH_videoroom_configure(self, request): try: videoroom_session = self.videoroom_sessions[request.session] except KeyError: raise APIError('Unknown video room session: {request.session}'.format(request=request, peer=self.protocol.peer)) videoroom = videoroom_session.room # todo: should we send out events if the active participant list did not change? try: videoroom.active_participants = request.active_participants except ValueError as e: raise APIError(str(e)) for session in videoroom: session.owner.send(sylkrtc.VideoroomConfigureEvent(session=session.id, active_participants=videoroom.active_participants, originator=request.session)) def _RH_videoroom_feed_attach(self, request): if request.feed in self.videoroom_sessions: raise APIError('Video room session ID {request.feed} already in use'.format(request=request, peer=self.protocol.peer)) try: base_session = self.videoroom_sessions[request.session] # our 'base' session (the one used to join and publish) except KeyError: raise APIError('Unknown video room session: {request.session}'.format(request=request, peer=self.protocol.peer)) try: publisher_session = base_session.room[request.publisher] # the publisher's session (the one we want to subscribe to) except KeyError: raise APIError('Unknown publisher video room session to attach to: {request.publisher}'.format(request=request, peer=self.protocol.peer)) if publisher_session.publisher_id is None: raise APIError('Video room session {session.id} does not have a publisher ID'.format(session=publisher_session)) videoroom_handle = VideoroomPluginHandle(self.janus_session, event_handler=self._handle_janus_videoroom_event) try: - videoroom_handle.feed_attach(room=base_session.room.id, feed=publisher_session.publisher_id) + videoroom_handle.feed_attach(room=base_session.room.id, feed=publisher_session.publisher_id, offer_audio=base_session.room.audio, offer_video=base_session.room.video) except Exception: videoroom_handle.detach() raise videoroom_session = VideoroomSessionInfo(request.feed, owner=self, janus_handle=videoroom_handle) videoroom_session.init_subscriber(publisher_session, parent_session=base_session) self.videoroom_sessions.add(videoroom_session) base_session.feeds.add(publisher_session) def _RH_videoroom_feed_answer(self, request): try: videoroom_session = self.videoroom_sessions[request.feed] except KeyError: raise APIError('Unknown video room session: {request.feed}'.format(request=request, peer=self.protocol.peer)) if videoroom_session.parent_session.id != request.session: raise APIError('{request.feed} is not an attached feed of {request.session}'.format(request=request, peer=self.protocol.peer)) videoroom_session.janus_handle.feed_start(sdp=request.sdp) def _RH_videoroom_feed_detach(self, request): try: videoroom_session = self.videoroom_sessions[request.feed] except KeyError: raise APIError('Unknown video room session to detach: {request.feed}'.format(request=request, peer=self.protocol.peer)) if videoroom_session.parent_session.id != request.session: raise APIError('{request.feed} is not an attached feed of {request.session}'.format(request=request, peer=self.protocol.peer)) videoroom_session.janus_handle.feed_detach() # safety net in case we do not get any answer for the feed_detach request # todo: to be adjusted later after pseudo-synchronous communication with janus is implemented reactor.callLater(2, call_in_green_thread, self._cleanup_videoroom_session, videoroom_session) def _RH_videoroom_invite(self, request): try: base_session = self.videoroom_sessions[request.session] except KeyError: raise APIError('Unknown video room session: {request.session}'.format(request=request, peer=self.protocol.peer)) room = base_session.room participants = set(request.participants) originator = sylkrtc.SIPIdentity(uri=base_session.account.id, display_name=base_session.account.display_name) session_id = str(random.getrandbits(32)) event = sylkrtc.AccountConferenceInviteEvent(account='placeholder', room=room.uri, originator=originator, session_id=self._callid_to_uuid(session_id)) for protocol in self.protocol.factory.connections.difference([self.protocol]): connection_handler = protocol.connection_handler for account in participants.intersection(connection_handler.accounts_map): event.account = account connection_handler.send(event) room.log.info('invitation from %s for %s with session-id %s', originator.uri, account, session_id) connection_handler.log.info('received an invitation from %s for %s to join video room %s', originator.uri, account, room.uri) for participant in participants: - push.conference_invite(originator=originator, destination=participant, room=room.uri, call_id=session_id) + push.conference_invite(originator=originator, destination=participant, room=room.uri, call_id=session_id, audio=room.audio, video=room.video) def _RH_videoroom_session_trickle(self, request): try: videoroom_session = self.videoroom_sessions[request.session] except KeyError: raise APIError('Unknown video room session: {request.session}'.format(request=request, peer=self.protocol.peer)) videoroom_session.janus_handle.trickle(request.candidates) if not request.candidates and videoroom_session.type == 'publisher': self.log.debug('video room session {session.id} negotiated ICE'.format(session=videoroom_session)) def _RH_videoroom_session_update(self, request): try: videoroom_session = self.videoroom_sessions[request.session] except KeyError: raise APIError('Unknown video room session: {request.session}'.format(request=request, peer=self.protocol.peer)) options = request.options.__data__ if options: videoroom_session.janus_handle.update_publisher(options) modified = ', '.join('{}={}'.format(key, options[key]) for key in options) self.log.info('updated video room session {request.session} with {modified}'.format(request=request, modified=modified)) def _RH_videoroom_message(self, request): try: videoroom_session = self.videoroom_sessions[request.session] except KeyError: raise APIError('Unknown video room session: {request.session}'.format(request=request, peer=self.protocol.peer)) content_type = request.content_type content = request.content if content_type.startswith('text') else request.content.encode('latin1') message_id = request.message_id.encode('ascii') videoroom_session.chat_handler.send_message(message_id, content, content_type) def _RH_videoroom_composing_indication(self, request): try: videoroom_session = self.videoroom_sessions[request.session] except KeyError: raise APIError('Unknown video room session: {request.session}'.format(request=request, peer=self.protocol.peer)) videoroom_session.chat_handler.send_composing_indication(request.state, request.refresh) def _RH_videoroom_mute_audio_participants(self, request): try: videoroom_session = self.videoroom_sessions[request.session] except KeyError: raise APIError('Unknown video room session: {request.session}'.format(request=request, peer=self.protocol.peer)) videoroom = videoroom_session.room for session in videoroom: session.owner.send(sylkrtc.VideoroomMuteAudioEvent(session=session.id, originator=request.session)) def _RH_videoroom_toggle_hand(self, request): try: videoroom_session = self.videoroom_sessions[request.session] except KeyError: raise APIError('Unknown video room session: {request.session}'.format(request=request, peer=self.protocol.peer)) videoroom = videoroom_session.room if request.session_id: request_session = request.session_id else: request_session = request.session videoroom.raised_hands = request_session for session in videoroom: session.owner.send(sylkrtc.VideoroomRaisedHandsEvent(session=session.id, raised_hands=videoroom.raised_hands)) # Event handlers def _EH_janus_sip(self, event): if isinstance(event, janus.PluginEvent): event_id = event.plugindata.data.__id__ try: handler = getattr(self, '_EH_janus_' + '_'.join(event_id)) except AttributeError: self.log.warning('unhandled Janus SIP event: {event_name}'.format(event_name=event_id[-1])) else: self.log.debug('janus SIP event: {event_name} (handle_id={event.sender})'.format(event=event, event_name=event_id[-1])) handler(event) else: # janus.CoreEvent try: handler = getattr(self, '_EH_janus_sip_' + event.janus) except AttributeError: self.log.warning('unhandled Janus SIP event: {event.janus}'.format(event=event)) else: self.log.debug('janus SIP event: {event.janus} (handle_id={event.sender})'.format(event=event)) handler(event) def _EH_janus_sip_error(self, event): # fixme: implement error handling self.log.error('got SIP error event: {}'.format(event.__data__)) handle_id = event.sender if handle_id in self.sip_sessions: pass # this is a session related event elif handle_id in self.account_handles_map: pass # this is an account related event def _EH_janus_sip_webrtcup(self, event): try: session_info = self.sip_sessions[event.sender] except KeyError: self.log.warning('could not find SIP session with handle ID {event.sender} for webrtcup event'.format(event=event)) return session_info.state = 'established' self.send(sylkrtc.SessionEstablishedEvent(session=session_info.id)) self.log.info('{peer} {session.direction} session {session.id} established'.format(session=session_info, peer=self.protocol.peer)) def _EH_janus_sip_hangup(self, event): try: session_info = self.sip_sessions[event.sender] except KeyError: return if session_info.state != 'terminated': session_info.state = 'terminated' reason = event.reason or 'unspecified reason' self.send(sylkrtc.SessionTerminatedEvent(session=session_info.id, reason=reason)) self.log.info('{peer} {session.direction} session {session.id} terminated ({reason})'.format(session=session_info, reason=reason, peer=self.protocol.peer)) self._cleanup_session(session_info) def _EH_janus_sip_slowlink(self, event): try: session_info = self.sip_sessions[event.sender] except KeyError: self.log.warning('could not find SIP session with handle ID {event.sender} for slowlink event'.format(event=event)) return if event.uplink: # uplink is from janus' point of view if not session_info.slow_download: self.log.debug('poor download connectivity for session {session.id}'.format(session=session_info, peer=self.protocol.peer)) session_info.slow_download = True else: if not session_info.slow_upload: self.log.debug('poor upload connectivity for session {session.id}'.format(session=session_info, peer=self.protocol.peer)) session_info.slow_upload = True def _EH_janus_sip_media(self, event): pass def _EH_janus_sip_detached(self, event): pass def _EH_janus_sip_event_registering(self, event): try: account_info = self.account_handles_map[event.sender] except KeyError: self.log.warning('could not find account with handle ID {event.sender} for registering event'.format(event=event)) return if account_info.registration_state != 'registering': account_info.registration_state = 'registering' self.send(sylkrtc.AccountRegisteringEvent(account=account_info.id)) def _EH_janus_sip_event_registered(self, event): if event.sender in self.sip_sessions: # skip 'registered' events from outgoing session handles return try: account_info = self.account_handles_map[event.sender] except KeyError: self.log.warning('could not find account with handle ID {event.sender} for registered event'.format(event=event)) return if account_info.registration_state != 'registered': account_info.registration_state = 'registered' self.send(sylkrtc.AccountRegisteredEvent(account=account_info.id)) self.log.info('{peer} registered account {account.id}'.format(account=account_info, peer=self.protocol.peer)) def _EH_janus_sip_event_registration_failed(self, event): try: account_info = self.account_handles_map[event.sender] except KeyError: self.log.warning('could not find account with handle ID {event.sender} for registration failed event'.format(event=event)) return if account_info.registration_state != 'failed': account_info.registration_state = 'failed' reason = '{result.code} {result.reason}'.format(result=event.plugindata.data.result) self.send(sylkrtc.AccountRegistrationFailedEvent(account=account_info.id, reason=reason)) self.log.info('{peer} register for {account.id} failed: {reason}'.format(account=account_info, reason=reason, peer=self.protocol.peer)) def _EH_janus_sip_event_incomingcall(self, event): try: account_info = self.account_handles_map[event.sender] except KeyError: self.log.warning('could not find account with handle ID {event.sender} for incoming call event'.format(event=event)) return assert event.jsep is not None data = event.plugindata.data.result # type: janus.SIPResultIncomingCall call_id = event.plugindata.data.call_id originator = sylkrtc.SIPIdentity(uri=data.username, display_name=data.displayname) session = SIPSessionInfo(self._callid_to_uuid(call_id)) session.janus_handle = account_info.janus_handle session.init_incoming(account_info, originator.uri, originator.display_name) self.sip_sessions.add(session) self.send(sylkrtc.AccountIncomingSessionEvent(account=account_info.id, session=session.id, originator=originator, sdp=event.jsep.sdp, call_id=call_id)) self.log.info('{peer} incoming session {session.id} from {session.remote_identity.uri!s} to {session.local_identity.uri!s}'.format(session=session, peer=self.protocol.peer)) def _EH_janus_sip_event_missed_call(self, event): try: account_info = self.account_handles_map[event.sender] except KeyError: self.log.warning('could not find account with handle ID {event.sender} for missed call event'.format(event=event)) return data = event.plugindata.data.result # type: janus.SIPResultMissedCall originator = sylkrtc.SIPIdentity(uri=data.caller, display_name=data.displayname) self.send(sylkrtc.AccountMissedSessionEvent(account=account_info.id, originator=originator)) self.log.info('missed incoming call from {originator.uri}'.format(originator=originator)) def _EH_janus_sip_event_calling(self, event): try: session_info = self.sip_sessions[event.sender] except KeyError: self.log.warning('could not find SIP session with handle ID {event.sender} for calling event'.format(event=event)) return session_info.state = 'progress' self.send(sylkrtc.SessionProgressEvent(session=session_info.id)) self.log.debug('{peer} {session.direction} session {session.id} state: {session.state}'.format(session=session_info, peer=self.protocol.peer)) def _EH_janus_sip_event_accepted(self, event): try: session_info = self.sip_sessions[event.sender] except KeyError: self.log.warning('could not find SIP session with handle ID {event.sender} for accepted event'.format(event=event)) return if session_info.state == 'established': # We had early media session_info.state = 'accepted' self.send(sylkrtc.SessionAcceptedEvent(session=session_info.id)) self.log.debug('{peer} {session.direction} session {session.id} state: {session.state}'.format(session=session_info, peer=self.protocol.peer)) return session_info.state = 'accepted' if session_info.direction == 'outgoing': assert event.jsep is not None self.send(sylkrtc.SessionAcceptedEvent(session=session_info.id, sdp=event.jsep.sdp, call_id=event.plugindata.data.call_id)) else: self.send(sylkrtc.SessionAcceptedEvent(session=session_info.id)) self.log.debug('{peer} {session.direction} session {session.id} state: {session.state}'.format(session=session_info, peer=self.protocol.peer)) def _EH_janus_sip_event_hangup(self, event): try: session_info = self.sip_sessions[event.sender] except KeyError: self.log.warning('could not find SIP session with handle ID {event.sender} for hangup event'.format(event=event)) return if session_info.state != 'terminated': session_info.state = 'terminated' data = event.plugindata.data.result # type: janus.SIPResultHangup reason = '{0.code} {0.reason}'.format(data) self.send(sylkrtc.SessionTerminatedEvent(session=session_info.id, reason=reason)) if session_info.direction == 'incoming' and data.code == 487: # incoming call was cancelled -> missed self.send(sylkrtc.AccountMissedSessionEvent(account=session_info.account.id, originator=session_info.remote_identity.__dict__)) if data.code >= 300: self.log.info('{peer} {session.direction} session {session.id} terminated ({reason})'.format(session=session_info, reason=reason, peer=self.protocol.peer)) else: self.log.info('{peer} {session.direction} session {session.id} terminated'.format(session=session_info, peer=self.protocol.peer)) self._cleanup_session(session_info) def _EH_janus_sip_event_declining(self, event): pass def _EH_janus_sip_event_hangingup(self, event): pass def _EH_janus_sip_event_proceeding(self, event): pass def _EH_janus_sip_event_progress(self, event): if (event.jsep): try: session_info = self.sip_sessions[event.sender] except KeyError: self.log.warning('could not find SIP session with handle ID {event.sender} for progress event'.format(event=event)) return session_info.state = 'early_media' self.log.info('{peer} {session.direction} session {session.id} has early media'.format(session=session_info, peer=self.protocol.peer)) self.send(sylkrtc.SessionEarlyMediaEvent(session=session_info.id, sdp=event.jsep.sdp, call_id=event.plugindata.data.call_id)) self.log.debug('{peer} {session.direction} session {session.id} state: {session.state}'.format(session=session_info, peer=self.protocol.peer)) def _EH_janus_sip_event_ringing(self, event): pass def _EH_janus_videoroom(self, event): if isinstance(event, janus.PluginEvent): event_id = event.plugindata.data.__id__ try: handler = getattr(self, '_EH_janus_' + '_'.join(event_id)) except AttributeError: self.log.warning('unhandled Janus videoroom event: {event_name}'.format(event_name=event_id[-1])) else: self.log.debug('janus videoroom event: {event_name} (handle_id={event.sender})'.format(event=event, event_name=event_id[-1])) handler(event) else: # janus.CoreEvent try: handler = getattr(self, '_EH_janus_videoroom_' + event.janus) except AttributeError: self.log.warning('unhandled Janus videoroom event: {event.janus}'.format(event=event)) else: self.log.debug('janus videoroom event: {event.janus} (handle_id={event.sender})'.format(event=event)) handler(event) def _EH_janus_videoroom_error(self, event): # fixme: implement error handling self.log.error('got videoroom error event: {}'.format(event.__data__)) try: videoroom_session = self.videoroom_sessions[event.sender] except KeyError: self.log.warning('could not find video room session with handle ID {event.sender} for error event'.format(event=event)) return if videoroom_session.type == 'publisher': pass else: pass def _EH_janus_videoroom_webrtcup(self, event): try: videoroom_session = self.videoroom_sessions[event.sender] except KeyError: self.log.warning('could not find video room session with handle ID {event.sender} for webrtcup event'.format(event=event)) return if videoroom_session.type == 'publisher': self.log.info('{peer} video room session {session.id} created'.format(session=videoroom_session, peer=self.protocol.peer)) self.send(sylkrtc.VideoroomSessionEstablishedEvent(session=videoroom_session.id)) else: self.send(sylkrtc.VideoroomFeedEstablishedEvent(session=videoroom_session.parent_session.id, feed=videoroom_session.id)) def _EH_janus_videoroom_hangup(self, event): try: videoroom_session = self.videoroom_sessions[event.sender] except KeyError: return self._cleanup_videoroom_session(videoroom_session) self.log.info('{peer} video room session {session.id} ended'.format(session=videoroom_session, peer=self.protocol.peer)) def _EH_janus_videoroom_slowlink(self, event): try: videoroom_session = self.videoroom_sessions[event.sender] except KeyError: self.log.warning('could not find video room session with handle ID {event.sender} for slowlink event'.format(event=event)) return if event.uplink: # uplink is from janus' point of view if not videoroom_session.slow_download: self.log.debug('poor download connectivity to video room {session.room.uri} with session {session.id}'.format(session=videoroom_session)) videoroom_session.slow_download = True else: if not videoroom_session.slow_upload: self.log.debug('poor upload connectivity to video room {session.room.uri} with session {session.id}'.format(session=videoroom_session)) videoroom_session.slow_upload = True def _EH_janus_videoroom_media(self, event): pass def _EH_janus_videoroom_detached(self, event): pass def _EH_janus_videoroom_joined(self, event): # send when a publisher successfully joined a room try: videoroom_session = self.videoroom_sessions[event.sender] except KeyError: self.log.warning('could not find video room session with handle ID {event.sender} for joined event'.format(event=event)) return self.log.info('{peer} joined video room {session.room.uri} with session {session.id}'.format(session=videoroom_session, peer=self.protocol.peer)) data = event.plugindata.data # type: janus.VideoroomJoined videoroom_session.publisher_id = data.id room = videoroom_session.room assert event.jsep is not None - self.send(sylkrtc.VideoroomSessionAcceptedEvent(session=videoroom_session.id, sdp=event.jsep.sdp)) + self.send(sylkrtc.VideoroomSessionAcceptedEvent(session=videoroom_session.id, sdp=event.jsep.sdp, audio=room.audio, video=room.video)) # send information about existing publishers publishers = [] for publisher in data.publishers: # type: janus.VideoroomPublisher try: publisher_session = room[publisher.id] except KeyError: self.log.warning('could not find matching session for publisher {publisher.id} during joined event'.format(publisher=publisher)) else: publishers.append(dict(id=publisher_session.id, uri=publisher_session.account.id, display_name=publisher.display or '')) self.send(sylkrtc.VideoroomInitialPublishersEvent(session=videoroom_session.id, publishers=publishers)) room.add(videoroom_session) # adding the session to the room might also trigger sending an event with the active participants which must be sent last def _EH_janus_videoroom_attached(self, event): # sent when a feed is subscribed for a given publisher try: videoroom_session = self.videoroom_sessions[event.sender] except KeyError: self.log.warning('could not find video room session with handle ID {event.sender} for attached event'.format(event=event)) return # get the session which originated the subscription base_session = videoroom_session.parent_session assert base_session is not None assert event.jsep is not None and event.jsep.type == 'offer' self.send(sylkrtc.VideoroomFeedAttachedEvent(session=base_session.id, feed=videoroom_session.id, sdp=event.jsep.sdp)) def _EH_janus_videoroom_slow_link(self, event): pass def _EH_janus_videoroom_event_publishers(self, event): try: videoroom_session = self.videoroom_sessions[event.sender] except KeyError: self.log.warning('could not find video room session with handle ID {event.sender} for publishers event'.format(event=event)) return room = videoroom_session.room # send information about new publishers publishers = [] for publisher in event.plugindata.data.publishers: # type: janus.VideoroomPublisher try: publisher_session = room[publisher.id] except KeyError: self.log.warning('could not find matching session for publisher {publisher.id} during publishers event'.format(publisher=publisher)) continue publishers.append(dict(id=publisher_session.id, uri=publisher_session.account.id, display_name=publisher.display or '')) self.send(sylkrtc.VideoroomPublishersJoinedEvent(session=videoroom_session.id, publishers=publishers)) def _EH_janus_videoroom_event_leaving(self, event): # this is a publisher publisher_id = event.plugindata.data.leaving # publisher_id == 'ok' when the event is about ourselves leaving the room, else the publisher's janus ID try: base_session = self.videoroom_sessions[event.sender] except KeyError: if publisher_id != 'ok': self.log.warning('could not find video room session with handle ID {event.sender} for leaving event'.format(event=event)) return if publisher_id == 'ok': self.log.info('{peer} left video room {session.room.uri} with session {session.id}'.format(session=base_session, peer=self.protocol.peer)) self._cleanup_videoroom_session(base_session) return try: publisher_session = base_session.feeds.pop(publisher_id) except KeyError: return self.send(sylkrtc.VideoroomPublishersLeftEvent(session=base_session.id, publishers=[publisher_session.id])) def _EH_janus_videoroom_event_left(self, event): # this is a subscriber try: videoroom_session = self.videoroom_sessions[event.sender] except KeyError: pass else: self._cleanup_videoroom_session(videoroom_session) def _EH_janus_videoroom_event_configured(self, event): pass def _EH_janus_videoroom_event_started(self, event): pass def _EH_janus_videoroom_event_unpublished(self, event): pass # Notification handlers def _NH_ChatSessionGotMessage(self, notification): session = notification.sender.sylk_session # type: VideoroomSessionInfo message = notification.data.message sender = sylkrtc.SIPIdentity(uri=str(message.sender.uri), display_name=message.sender.display_name) content = message.content if isinstance(message.content, unicode) else message.content.decode('latin1') # preserve binary data for transmitting over JSON if any(header.name == 'Message-Type' and header.value == 'status' and header.namespace == 'urn:ag-projects:xml:ns:cpim' for header in message.additional_headers): message_type = 'status' else: message_type = 'normal' self.send(sylkrtc.VideoroomMessageEvent(session=session.id, content=content, content_type=message.content_type, sender=sender, timestamp=str(message.timestamp), type=message_type)) def _NH_ChatSessionGotComposingIndication(self, notification): session = notification.sender.sylk_session # type: VideoroomSessionInfo composing = notification.data sender = sylkrtc.SIPIdentity(uri=str(composing.sender.uri), display_name=composing.sender.display_name) self.send(sylkrtc.VideoroomComposingIndicationEvent(session=session.id, state=composing.state, refresh=composing.refresh, content_type=composing.content_type, sender=sender)) def _NH_ChatSessionDidDeliverMessage(self, notification): session = notification.sender.sylk_session # type: VideoroomSessionInfo data = notification.data self.send(sylkrtc.VideoroomMessageDeliveryEvent(session=session.id, delivered=True, message_id=data.message_id, code=data.code, reason=data.reason)) def _NH_ChatSessionDidNotDeliverMessage(self, notification): session = notification.sender.sylk_session # type: VideoroomSessionInfo data = notification.data self.send(sylkrtc.VideoroomMessageDeliveryEvent(session=session.id, delivered=False, message_id=data.message_id, code=data.code, reason=data.reason)) # noinspection PyPep8Naming class VideoroomChatHandler(object): implements(IObserver) def __init__(self, session): self.sylk_session = session # type: VideoroomSessionInfo self.sip_session = None # type: Optional[Session] self.chat_stream = None self._started = False self._ended = False self._message_queue = deque() @property def account(self): return self.sylk_session.account @property def room(self): return self.sylk_session.room @run_in_green_thread def start(self): if self._started: return self._started = True notification_center = NotificationCenter() from_uri = SIPURI.parse(self.account.uri) to_uri = SIPURI.parse('sip:{}'.format(self.room.uri)) to_uri.host = to_uri.host.replace('videoconference', 'conference', 1) # TODO: find a way to define this credentials = Credentials(username=from_uri.user, password=self.account.password.encode('utf-8'), digest=True) sip_account = DefaultAccount() sip_settings = SIPSimpleSettings() if sip_account.sip.outbound_proxy is not None: uri = SIPURI(host=sip_account.sip.outbound_proxy.host, port=sip_account.sip.outbound_proxy.port, parameters={'transport': sip_account.sip.outbound_proxy.transport}) else: uri = to_uri lookup = DNSLookup() try: route = lookup.lookup_sip_proxy(uri, sip_settings.sip.transport_list).wait()[0] except (DNSLookupError, IndexError): self.end() self.room.log.error('DNS lookup for SIP proxy for {} failed'.format(uri)) self.room.log.error('chatroom session for {} failed: DNS lookup error'.format(self.account.id)) notification_center.post_notification('ChatSessionDidFail', sender=self, data=NotificationData(originator='local', code=0, reason=None, failure_reason='DNS lookup error')) return if self._ended: # end was called during DNS lookup self.room.log.info('chatroom session for {} ended'.format(self.account.id)) notification_center.post_notification('ChatSessionDidEnd', sender=self) return self.sip_session = Session(sip_account) self.chat_stream = MediaStreamRegistry.ChatStream() notification_center.add_observer(self, sender=self.sip_session) notification_center.add_observer(self, sender=self.chat_stream) self.sip_session.connect(FromHeader(from_uri, self.account.display_name), ToHeader(to_uri), route=route, streams=[self.chat_stream], credentials=credentials) @run_in_twisted_thread def end(self): if self._ended: return notification_center = NotificationCenter() if self.sip_session is not None: notification_center.remove_observer(self, sender=self.sip_session) notification_center.remove_observer(self, sender=self.chat_stream) self.sip_session.end() self.sip_session = None self.chat_stream = None self.room.log.info('chatroom session for {} ended'.format(self.account.id)) notification_center.post_notification('ChatSessionDidEnd', sender=self) while self._message_queue: message_id, content, content_type = self._message_queue.popleft() data = NotificationData(message_id=message_id, message=None, code=0, reason='Chat session ended') notification_center.post_notification('ChatSessionDidNotDeliverMessage', sender=self, data=data) self._ended = True @run_in_twisted_thread def send_message(self, message_id, content, content_type='text/plain'): if self._ended: notification_center = NotificationCenter() data = NotificationData(message_id=message_id, message=None, code=0, reason='Chat session ended') notification_center.post_notification('ChatSessionDidNotDeliverMessage', sender=self, data=data) else: self._message_queue.append((message_id, content, content_type)) if self.chat_stream is not None: self._send_queued_messages() @run_in_twisted_thread def send_composing_indication(self, state, refresh=None): if self.chat_stream is not None: self.chat_stream.send_composing_indication(state, refresh=refresh) def _send_queued_messages(self): while self._message_queue: message_id, content, content_type = self._message_queue.popleft() self.chat_stream.send_message(content, content_type, message_id=message_id) def handle_notification(self, notification): handler = getattr(self, '_NH_%s' % notification.name, Null) handler(notification) def _NH_SIPSessionDidStart(self, notification): self.room.log.info('chatroom session for {} started'.format(self.account.id)) notification.center.post_notification('ChatSessionDidStart', sender=self) self._send_queued_messages() def _NH_SIPSessionDidEnd(self, notification): notification.center.remove_observer(self, sender=self.sip_session) notification.center.remove_observer(self, sender=self.chat_stream) self.sip_session = None self.chat_stream = None self.end() self.room.log.info('chatroom session for {} ended'.format(self.account.id)) notification.center.post_notification('ChatSessionDidEnd', sender=self, data=notification.data) def _NH_SIPSessionDidFail(self, notification): notification.center.remove_observer(self, sender=self.sip_session) notification.center.remove_observer(self, sender=self.chat_stream) self.sip_session = None self.chat_stream = None self.end() self.room.log.error('chatroom session for {} failed: {}'.format(self.account.id, notification.data.failure_reason)) notification.center.post_notification('ChatSessionDidFail', sender=self, data=notification.data) # noinspection PyUnusedLocal def _NH_SIPSessionNewProposal(self, notification): self.sip_session.reject_proposal() def _NH_SIPSessionTransferNewIncoming(self, notification): # sylkserver's SIP Session class doesn't implement the transfer API # self.sip_session.reject_transfer(403) pass def _NH_ChatStreamGotMessage(self, notification): self.chat_stream.msrp_session.send_report(notification.data.chunk, 200, 'OK') notification.center.post_notification('ChatSessionGotMessage', sender=self, data=notification.data) def _NH_ChatStreamGotComposingIndication(self, notification): notification.center.post_notification('ChatSessionGotComposingIndication', sender=self, data=notification.data) def _NH_ChatStreamDidSendMessage(self, notification): notification.center.post_notification('ChatSessionDidSendMessage', sender=self, data=notification.data) def _NH_ChatStreamDidDeliverMessage(self, notification): notification.center.post_notification('ChatSessionDidDeliverMessage', sender=self, data=notification.data) def _NH_ChatStreamDidNotDeliverMessage(self, notification): notification.center.post_notification('ChatSessionDidNotDeliverMessage', sender=self, data=notification.data) diff --git a/sylk/applications/webrtcgateway/janus.py b/sylk/applications/webrtcgateway/janus.py index 318df37..6888240 100644 --- a/sylk/applications/webrtcgateway/janus.py +++ b/sylk/applications/webrtcgateway/janus.py @@ -1,357 +1,357 @@ import json from application.notification import IObserver, NotificationCenter, NotificationData from application.python import Null from application.python.types import Singleton from autobahn.twisted.websocket import connectWS, WebSocketClientFactory, WebSocketClientProtocol from eventlib.twistedutil import block_on from twisted.internet import reactor, defer from twisted.internet.protocol import ReconnectingClientFactory from twisted.python.failure import Failure from zope.interface import implements from sylk import __version__ from .configuration import JanusConfig from .logger import log from .models import janus class JanusError(Exception): def __init__(self, code, reason): super(JanusError, self).__init__(reason) self.code = code self.reason = reason class JanusClientProtocol(WebSocketClientProtocol): _event_handlers = None _pending_transactions = None _keepalive_timers = None _keepalive_interval = 45 notification_center = NotificationCenter() def onOpen(self): self.notification_center.post_notification('JanusBackendConnected', sender=self) self._pending_transactions = {} self._keepalive_timers = {} self._event_handlers = {} def onMessage(self, payload, isBinary): if isBinary: log.warn('Unexpected binary payload received') return self.notification_center.post_notification('WebRTCJanusTrace', sender=self, data=NotificationData(direction='INCOMING', message=payload, peer=self.peer)) try: message = janus.JanusMessage.from_payload(json.loads(payload)) except Exception as e: log.warning('Error decoding Janus message: {!s}'.format(e)) return if isinstance(message, (janus.CoreEvent, janus.PluginEvent)): # some of the plugin events might have the transaction, but we do not finalize # the transaction for them as they are not direct responses for the transaction handler = self._event_handlers.get(message.sender, Null) try: handler(message) except Exception as e: log.exception('Error while running Janus event handler: {!s}'.format(e)) return # at this point it can only be a response. clear the transaction and return the answer. try: request, deferred = self._pending_transactions.pop(message.transaction) except KeyError: log.warn('Discarding unexpected response: %s' % payload) return if isinstance(message, janus.AckResponse): deferred.callback(None) elif isinstance(message, janus.SuccessResponse): deferred.callback(message) elif isinstance(message, janus.ErrorResponse): deferred.errback(JanusError(message.error.code, message.error.reason)) else: assert isinstance(message, janus.PluginResponse) plugin_data = message.plugindata.data if isinstance(plugin_data, (janus.SIPErrorEvent, janus.VideoroomErrorEvent)): deferred.errback(JanusError(plugin_data.error_code, plugin_data.error)) else: deferred.callback(message) def connectionLost(self, reason): super(JanusClientProtocol, self).connectionLost(reason) self.notification_center.post_notification('JanusBackendDisconnected', sender=self, data=NotificationData(reason=reason.getErrorMessage())) def disconnect(self, code=1000, reason=u''): self.sendClose(code, reason) def _send_request(self, request): if request.janus != 'keepalive' and 'session_id' in request: # postpone keepalive messages as long as we have non-keepalive traffic for a given session keepalive_timer = self._keepalive_timers.get(request.session_id, None) if keepalive_timer is not None and keepalive_timer.active(): keepalive_timer.reset(self._keepalive_interval) deferred = defer.Deferred() message = json.dumps(request.__data__) self.notification_center.post_notification('WebRTCJanusTrace', sender=self, data=NotificationData(direction='OUTGOING', message=message, peer=self.peer)) self.sendMessage(message) self._pending_transactions[request.transaction] = request, deferred return deferred def _start_keepalive(self, response): session_id = response.data.id self._keepalive_timers[session_id] = reactor.callLater(self._keepalive_interval, self._send_keepalive, session_id) return response def _stop_keepalive(self, session_id): timer = self._keepalive_timers.pop(session_id, None) if timer is not None and timer.active(): timer.cancel() def _send_keepalive(self, session_id): deferred = self._send_request(janus.SessionKeepaliveRequest(session_id=session_id)) deferred.addBoth(self._keepalive_callback, session_id) def _keepalive_callback(self, result, session_id): if isinstance(result, Failure): self._keepalive_timers.pop(session_id) else: self._keepalive_timers[session_id] = reactor.callLater(self._keepalive_interval, self._send_keepalive, session_id) # Public API def set_event_handler(self, handle_id, event_handler): if event_handler is None: self._event_handlers.pop(handle_id, None) else: assert callable(event_handler) self._event_handlers[handle_id] = event_handler def info(self): return self._send_request(janus.InfoRequest()) def create_session(self): return self._send_request(janus.SessionCreateRequest()).addCallback(self._start_keepalive) def destroy_session(self, session_id): self._stop_keepalive(session_id) return self._send_request(janus.SessionDestroyRequest(session_id=session_id)) def attach_plugin(self, session_id, plugin): return self._send_request(janus.PluginAttachRequest(session_id=session_id, plugin=plugin)) def detach_plugin(self, session_id, handle_id): return self._send_request(janus.PluginDetachRequest(session_id=session_id, handle_id=handle_id)) def message(self, session_id, handle_id, body, jsep=None): if jsep is not None: return self._send_request(janus.MessageRequest(session_id=session_id, handle_id=handle_id, body=body, jsep=jsep)) else: return self._send_request(janus.MessageRequest(session_id=session_id, handle_id=handle_id, body=body)) def trickle(self, session_id, handle_id, candidates): return self._send_request(janus.TrickleRequest(session_id=session_id, handle_id=handle_id, candidates=candidates)) class JanusClientFactory(ReconnectingClientFactory, WebSocketClientFactory): noisy = False protocol = JanusClientProtocol class JanusBackend(object): __metaclass__ = Singleton implements(IObserver) def __init__(self): self.factory = JanusClientFactory(url=JanusConfig.api_url, protocols=['janus-protocol'], useragent='SylkServer/%s' % __version__) self.connector = None self.protocol = Null self._stopped = False @property def ready(self): return self.protocol is not Null def start(self): notification_center = NotificationCenter() notification_center.add_observer(self, name='JanusBackendConnected') notification_center.add_observer(self, name='JanusBackendDisconnected') self.connector = connectWS(self.factory) def stop(self): if self._stopped: return self._stopped = True self.factory.stopTrying() notification_center = NotificationCenter() notification_center.discard_observer(self, name='JanusBackendConnected') notification_center.discard_observer(self, name='JanusBackendDisconnected') if self.connector is not None: self.connector.disconnect() self.connector = None if self.protocol is not None: self.protocol.disconnect() self.protocol = Null def set_event_handler(self, handle_id, event_handler): self.protocol.set_event_handler(handle_id, event_handler) def info(self): return self.protocol.info() def create_session(self): return self.protocol.create_session() def destroy_session(self, session_id): return self.protocol.destroy_session(session_id) def attach_plugin(self, session_id, plugin): return self.protocol.attach_plugin(session_id, plugin) def detach_plugin(self, session_id, handle_id): return self.protocol.detach_plugin(session_id, handle_id) def message(self, session_id, handle_id, body, jsep=None): return self.protocol.message(session_id, handle_id, body, jsep) def trickle(self, session_id, handle_id, candidates): return self.protocol.trickle(session_id, handle_id, candidates) # Notification handling def handle_notification(self, notification): handler = getattr(self, '_NH_%s' % notification.name, Null) handler(notification) def _NH_JanusBackendConnected(self, notification): assert self.protocol is Null self.protocol = notification.sender log.info('Janus backend connection up') self.factory.resetDelay() def _NH_JanusBackendDisconnected(self, notification): log.info('Janus backend connection down: %s' % notification.data.reason) self.protocol = Null class JanusSession(object): backend = JanusBackend() def __init__(self): response = block_on(self.backend.create_session()) # type: janus.SuccessResponse self.id = response.data.id def destroy(self): return self.backend.destroy_session(self.id) class JanusPluginHandle(object): backend = JanusBackend() plugin = None def __init__(self, session, event_handler): if self.plugin is None: raise TypeError('Cannot instantiate {0.__class__.__name__} with no associated plugin'.format(self)) response = block_on(self.backend.attach_plugin(session.id, self.plugin)) # type: janus.SuccessResponse self.id = response.data.id self.session = session self.backend.set_event_handler(self.id, event_handler) def __enter__(self): return self def __exit__(self, exception_type, exception_value, traceback): self.detach() def detach(self): try: block_on(self.backend.detach_plugin(self.session.id, self.id)) except JanusError as e: log.warning('could not detach Janus plugin: %s', e) self.backend.set_event_handler(self.id, None) def message(self, body, jsep=None, async=False): deferred = self.backend.message(self.session.id, self.id, body, jsep) return deferred if async else block_on(deferred) def trickle(self, candidates, async=False): deferred = self.backend.trickle(self.session.id, self.id, candidates) return deferred if async else block_on(deferred) class GenericPluginHandle(JanusPluginHandle): def __init__(self, plugin, session, event_handler): self.plugin = plugin super(GenericPluginHandle, self).__init__(session, event_handler) class SIPPluginHandle(JanusPluginHandle): plugin = 'janus.plugin.sip' def register(self, account, proxy=None): send_register = True if account.auth_handle.type == 'SIP' else False self.message(janus.SIPRegister(proxy=proxy, send_register=send_register, **account.user_data)) def unregister(self): self.message(janus.SIPUnregister()) def call(self, account, uri, sdp, proxy=None): # in order to make a call we need to register first. do so without actually registering, as we are already registered self.message(janus.SIPRegister(proxy=proxy, send_register=False, **account.user_data)) self.message(janus.SIPCall(uri=uri, srtp='sdes_optional'), jsep=janus.SDPOffer(sdp=sdp)) def accept(self, sdp): self.message(janus.SIPAccept(), jsep=janus.SDPAnswer(sdp=sdp)) def decline(self, code=486): self.message(janus.SIPDecline(code=code)) def hangup(self): self.message(janus.SIPHangup()) class VideoroomPluginHandle(JanusPluginHandle): plugin = 'janus.plugin.videoroom' def create(self, room, config, publishers=10): self.message(janus.VideoroomCreate(room=room, publishers=publishers, **config.janus_data)) def destroy(self, room): try: self.message(janus.VideoroomDestroy(room=room)) except JanusError as e: log.warning('could not destroy video room %s: %s', room, e) - def join(self, room, sdp, display_name=None): + def join(self, room, sdp, audio, video, display_name=None): if display_name: - self.message(janus.VideoroomJoin(room=room, display=display_name), jsep=janus.SDPOffer(sdp=sdp)) + self.message(janus.VideoroomJoin(room=room, audio=audio, video=video, display=display_name), jsep=janus.SDPOffer(sdp=sdp)) else: - self.message(janus.VideoroomJoin(room=room), jsep=janus.SDPOffer(sdp=sdp)) + self.message(janus.VideoroomJoin(room=room, audio=audio, video=video), jsep=janus.SDPOffer(sdp=sdp)) def leave(self): self.message(janus.VideoroomLeave()) def update_publisher(self, options): self.message(janus.VideoroomUpdatePublisher(**options)) - def feed_attach(self, room, feed): - self.message(janus.VideoroomFeedAttach(room=room, feed=feed)) + def feed_attach(self, room, feed, offer_audio, offer_video): + self.message(janus.VideoroomFeedAttach(room=room, feed=feed, offer_audio=offer_audio, offer_video=offer_video)) def feed_detach(self): self.message(janus.VideoroomFeedDetach()) def feed_start(self, sdp): self.message(janus.VideoroomFeedStart(), jsep=janus.SDPAnswer(sdp=sdp)) def feed_pause(self): self.message(janus.VideoroomFeedPause()) def feed_resume(self): self.message(janus.VideoroomFeedStart()) def feed_update(self, options): self.message(janus.VideoroomFeedUpdate(**options)) diff --git a/sylk/applications/webrtcgateway/models/janus.py b/sylk/applications/webrtcgateway/models/janus.py index fb7da5d..8f4f21f 100644 --- a/sylk/applications/webrtcgateway/models/janus.py +++ b/sylk/applications/webrtcgateway/models/janus.py @@ -1,886 +1,888 @@ import os from application.python import subclasses from application.python.descriptor import classproperty from binascii import b2a_hex as hex_encode from typing import Union from ..configuration import JanusConfig from .jsonobjects import AbstractProperty, BooleanProperty, IntegerProperty, StringProperty, ArrayProperty, ObjectProperty from .jsonobjects import FixedValueProperty, LimitedChoiceProperty, AbstractObjectProperty, JSONObject, JSONArray from .sylkrtc import ICECandidates from .validators import URIValidator # Base models (these are abstract and should not be used directly) class JanusRequestBase(JSONObject): transaction = StringProperty() apisecret = FixedValueProperty(JanusConfig.api_secret or None) def __init__(self, **kw): if 'transaction' not in kw: kw['transaction'] = hex_encode(os.urandom(16)) # uuid4().hex is really slow super(JanusRequestBase, self).__init__(**kw) class JanusCoreMessageBase(JSONObject): # base class for messages/ sent by the Janus core (responses and events) pass class JanusPluginMessageBase(JSONObject): # base class for messages sent by the Janus plugins (responses and events) pass class PluginDataBase(JSONObject): pass class SIPPluginData(PluginDataBase): __plugin__ = 'sip' # noinspection PyMethodParameters,PyUnresolvedReferences @classproperty def __id__(cls): if cls is SIPErrorEvent: return cls.__plugin__, cls.sip.value, cls.error.name else: return cls.__plugin__, cls.sip.value, cls.result.object_type.event.value class VideoroomPluginData(PluginDataBase): __plugin__ = 'videoroom' __event__ = None # for events, the name of the property that describes the event # noinspection PyMethodParameters,PyUnresolvedReferences @classproperty def __id__(cls): if cls.__event__ is not None: return cls.__plugin__, cls.videoroom.value, cls.__event__ else: return cls.__plugin__, cls.videoroom.value class CoreResponse(JanusCoreMessageBase): transaction = StringProperty() class CoreEvent(JanusCoreMessageBase): session_id = IntegerProperty() sender = IntegerProperty() # Miscellaneous models class NumericId(JSONObject): id = IntegerProperty() class UserId(JSONObject): id = IntegerProperty() display = StringProperty(optional=True) class ErrorInfo(JSONObject): code = IntegerProperty() reason = StringProperty() class PluginDataContainer(JSONObject): plugin = StringProperty() data = AbstractObjectProperty() # type: Union[SIPPluginData, VideoroomPluginData] class SIPDataContainer(JSONObject): sip = FixedValueProperty('event') result = AbstractObjectProperty() class SDPOffer(JSONObject): type = FixedValueProperty('offer') sdp = StringProperty() class SDPAnswer(JSONObject): type = FixedValueProperty('answer') sdp = StringProperty() class JSEP(JSONObject): type = LimitedChoiceProperty(['offer', 'answer']) sdp = StringProperty() class VideoroomPublisher(JSONObject): id = IntegerProperty() display = StringProperty(optional=True) audio_codec = StringProperty(optional=True) video_codec = StringProperty(optional=True) talking = BooleanProperty(optional=True) class VideoroomPublishers(JSONArray): item_type = VideoroomPublisher class ContactParams(JSONObject): pn_app = StringProperty(optional=True) pn_tok = StringProperty(optional=True) pn_type = StringProperty(optional=True) pn_silent = StringProperty(optional=True) pn_device = StringProperty(optional=True) @property def __data__(self): data = super(ContactParams, self).__data__ for key in data: data[key.replace('_', '-')] = data.pop(key) return data # Janus requests class InfoRequest(JanusRequestBase): janus = FixedValueProperty('info') class SessionCreateRequest(JanusRequestBase): janus = FixedValueProperty('create') class SessionDestroyRequest(JanusRequestBase): janus = FixedValueProperty('destroy') session_id = IntegerProperty() class SessionKeepaliveRequest(JanusRequestBase): janus = FixedValueProperty('keepalive') session_id = IntegerProperty() class PluginAttachRequest(JanusRequestBase): janus = FixedValueProperty('attach') session_id = IntegerProperty() plugin = StringProperty() class PluginDetachRequest(JanusRequestBase): janus = FixedValueProperty('detach') session_id = IntegerProperty() handle_id = IntegerProperty() class MessageRequest(JanusRequestBase): janus = FixedValueProperty('message') session_id = IntegerProperty() handle_id = IntegerProperty() body = AbstractObjectProperty() jsep = AbstractObjectProperty(optional=True) class TrickleRequest(JanusRequestBase): janus = FixedValueProperty('trickle') session_id = IntegerProperty() handle_id = IntegerProperty() candidates = ArrayProperty(ICECandidates) # type: ICECandidates @property def __data__(self): data = super(TrickleRequest, self).__data__ candidates = data.pop('candidates', []) if len(candidates) == 0: data['candidate'] = {'completed': True} # data['candidate'] = None elif len(candidates) == 1: data['candidate'] = candidates[0] else: data['candidates'] = candidates return data # SIP plugin messages (to be used as body for MessageRequest for the SIP plugin messages) class SIPRegister(JSONObject): request = FixedValueProperty('register') username = StringProperty() # this is the account.uri which is already a valid URI ha1_secret = StringProperty() display_name = StringProperty(optional=True) user_agent = StringProperty(optional=True) proxy = StringProperty(optional=True) send_register = BooleanProperty(optional=True) contact_params = ObjectProperty(ContactParams) class SIPUnregister(JSONObject): request = FixedValueProperty('unregister') class SIPCall(JSONObject): request = FixedValueProperty('call') uri = StringProperty(validator=URIValidator()) # this comes from the client request.uri which was validated as an AOR and we need a URI srtp = LimitedChoiceProperty(['sdes_optional', 'sdes_mandatory'], optional=True) class SIPAccept(JSONObject): request = FixedValueProperty('accept') class SIPDecline(JSONObject): request = FixedValueProperty('decline') code = IntegerProperty() class SIPHangup(JSONObject): request = FixedValueProperty('hangup') # Videoroom plugin messages (to be used as body for MessageRequest for videoroom plugin messages) class VideoroomCreate(JSONObject): request = FixedValueProperty('create') room = IntegerProperty() pin = StringProperty(optional=True) description = StringProperty(optional=True) publishers = IntegerProperty(optional=True, default=10) videocodec = StringProperty(optional=True) # don't need a validator, as the value comes from the configuration and it's validated there bitrate = IntegerProperty(optional=True) # don't need a validator, as the value comes from the configuration and it's validated there record = BooleanProperty(optional=True, default=False) rec_dir = StringProperty(optional=True) class VideoroomDestroy(JSONObject): request = FixedValueProperty('destroy') room = IntegerProperty() class VideoroomJoin(JSONObject): request = FixedValueProperty('joinandconfigure') ptype = FixedValueProperty('publisher') room = IntegerProperty() display = StringProperty(optional=True) audio = BooleanProperty(optional=True, default=True) video = BooleanProperty(optional=True, default=True) bitrate = IntegerProperty(optional=True) class VideoroomLeave(JSONObject): request = FixedValueProperty('leave') class VideoroomUpdatePublisher(JSONObject): request = FixedValueProperty('configure') audio = BooleanProperty(optional=True) video = BooleanProperty(optional=True) bitrate = IntegerProperty(optional=True) class VideoroomFeedAttach(JSONObject): request = FixedValueProperty('join') ptype = FixedValueProperty('subscriber') room = IntegerProperty() feed = IntegerProperty() + offer_audio = BooleanProperty(optional=True, default=True) + offer_video = BooleanProperty(optional=True, default=True) class VideoroomFeedDetach(JSONObject): request = FixedValueProperty('leave') class VideoroomFeedStart(JSONObject): request = FixedValueProperty('start') class VideoroomFeedPause(JSONObject): request = FixedValueProperty('pause') class VideoroomFeedUpdate(JSONObject): request = FixedValueProperty('configure') audio = BooleanProperty(optional=True) video = BooleanProperty(optional=True) substream = IntegerProperty(optional=True) # for VP8 simulcast temporal = IntegerProperty(optional=True) # for VP8 simulcast spatial_layer = IntegerProperty(optional=True) # for VP9 SVC temporal_layer = IntegerProperty(optional=True) # for VP9 SVC # Janus core messages class AckResponse(CoreResponse): janus = FixedValueProperty('ack') session_id = IntegerProperty() class ErrorResponse(CoreResponse): janus = FixedValueProperty('error') session_id = IntegerProperty(optional=True) # not used error = ObjectProperty(ErrorInfo) # type: ErrorInfo class SuccessResponse(CoreResponse): janus = FixedValueProperty('success') session_id = IntegerProperty(optional=True) # not used data = ObjectProperty(NumericId, optional=True) # type: NumericId class ServerInfoResponse(CoreResponse): janus = FixedValueProperty('server_info') name = StringProperty() version = IntegerProperty() version_string = StringProperty() data_channels = BooleanProperty() class DetachedEvent(CoreEvent): janus = FixedValueProperty('detached') class HangupEvent(CoreEvent): janus = FixedValueProperty('hangup') reason = StringProperty(optional=True) class MediaEvent(CoreEvent): janus = FixedValueProperty('media') type = LimitedChoiceProperty(['audio', 'video']) receiving = BooleanProperty() seconds = IntegerProperty(optional=True) class SlowlinkEvent(CoreEvent): janus = FixedValueProperty('slowlink') media = LimitedChoiceProperty(['audio', 'video']) uplink = BooleanProperty() lost = IntegerProperty() class WebrtcUpEvent(CoreEvent): janus = FixedValueProperty('webrtcup') # Janus plugin messages class PluginResponse(JanusPluginMessageBase): janus = FixedValueProperty('success') transaction = StringProperty() session_id = IntegerProperty() sender = IntegerProperty() plugindata = ObjectProperty(PluginDataContainer) # type: PluginDataContainer class PluginEvent(JanusPluginMessageBase): janus = FixedValueProperty('event') transaction = StringProperty(optional=True) session_id = IntegerProperty() sender = IntegerProperty() plugindata = ObjectProperty(PluginDataContainer) # type: PluginDataContainer jsep = ObjectProperty(JSEP, optional=True) # type: JSEP # SIP plugin data messages # Results class SIPResultRegistering(JSONObject): event = FixedValueProperty('registering') class SIPResultRegistered(JSONObject): event = FixedValueProperty('registered') username = StringProperty() # not used register_sent = BooleanProperty() class SIPResultRegistrationFailed(JSONObject): event = FixedValueProperty('registration_failed') code = IntegerProperty() reason = StringProperty() class SIPResultUnregistering(JSONObject): event = FixedValueProperty('unregistering') class SIPResultUnregistered(JSONObject): event = FixedValueProperty('unregistered') username = StringProperty() # not used class SIPResultCalling(JSONObject): event = FixedValueProperty('calling') class SIPResultRinging(JSONObject): event = FixedValueProperty('ringing') class SIPResultProceeding(JSONObject): event = FixedValueProperty('proceeding') code = IntegerProperty() class SIPResultProgress(JSONObject): event = FixedValueProperty('progress') username = StringProperty() # not used class SIPResultDeclining(JSONObject): event = FixedValueProperty('declining') code = IntegerProperty() class SIPResultAccepting(JSONObject): event = FixedValueProperty('accepting') class SIPResultAccepted(JSONObject): event = FixedValueProperty('accepted') username = StringProperty(optional=True) # not used (only present for outgoing) class SIPResultHolding(JSONObject): event = FixedValueProperty('holding') class SIPResultResuming(JSONObject): event = FixedValueProperty('resuming') class SIPResultHangingUp(JSONObject): event = FixedValueProperty('hangingup') class SIPResultHangup(JSONObject): event = FixedValueProperty('hangup') code = IntegerProperty() reason = StringProperty() class SIPResultIncomingCall(JSONObject): event = FixedValueProperty('incomingcall') username = StringProperty() displayname = StringProperty(optional=True) srtp = LimitedChoiceProperty(['sdes_optional', 'sdes_mandatory'], optional=True) class SIPResultMissedCall(JSONObject): event = FixedValueProperty('missed_call') caller = StringProperty() displayname = StringProperty(optional=True) class SIPResultInfo(JSONObject): event = FixedValueProperty('info') sender = StringProperty() displayname = StringProperty(optional=True) type = StringProperty() content = StringProperty() class SIPResultInfoSent(JSONObject): event = FixedValueProperty('infosent') class SIPResultMessage(JSONObject): event = FixedValueProperty('message') sender = StringProperty() displayname = StringProperty(optional=True) content = StringProperty() class SIPResultMessageSent(JSONObject): event = FixedValueProperty('messagesent') class SIPResultDTMFSent(JSONObject): event = FixedValueProperty('dtmfsent') # Data messages class SIPErrorEvent(SIPPluginData): sip = FixedValueProperty('event') error_code = IntegerProperty() error = StringProperty() class SIPRegisteringEvent(SIPPluginData): sip = FixedValueProperty('event') result = ObjectProperty(SIPResultRegistering) # type: SIPResultRegistering class SIPRegisteredEvent(SIPPluginData): sip = FixedValueProperty('event') result = ObjectProperty(SIPResultRegistered) # type: SIPResultRegistered class SIPRegistrationFailedEvent(SIPPluginData): sip = FixedValueProperty('event') result = ObjectProperty(SIPResultRegistrationFailed) # type: SIPResultRegistrationFailed class SIPUnregisteringEvent(SIPPluginData): sip = FixedValueProperty('event') result = ObjectProperty(SIPResultUnregistering) # type: SIPResultUnregistering class SIPUnregisteredEvent(SIPPluginData): sip = FixedValueProperty('event') result = ObjectProperty(SIPResultUnregistered) # type: SIPResultRegistered class SIPCallingEvent(SIPPluginData): sip = FixedValueProperty('event') result = ObjectProperty(SIPResultCalling) # type: SIPResultCalling class SIPRingingEvent(SIPPluginData): sip = FixedValueProperty('event') result = ObjectProperty(SIPResultRinging) # type: SIPResultRinging class SIPProceedingEvent(SIPPluginData): sip = FixedValueProperty('event') result = ObjectProperty(SIPResultProceeding) # type: SIPResultProceeding class SIPProgressEvent(SIPPluginData): sip = FixedValueProperty('event') result = ObjectProperty(SIPResultProgress) # type: SIPResultProgress call_id = StringProperty(optional=True) class SIPDecliningEvent(SIPPluginData): sip = FixedValueProperty('event') result = ObjectProperty(SIPResultDeclining) # type: SIPResultDeclining class SIPAcceptingEvent(SIPPluginData): sip = FixedValueProperty('event') result = ObjectProperty(SIPResultAccepting) # type: SIPResultAccepting class SIPAcceptedEvent(SIPPluginData): sip = FixedValueProperty('event') result = ObjectProperty(SIPResultAccepted) # type: SIPResultAccepted call_id = StringProperty() class SIPHoldingEvent(SIPPluginData): sip = FixedValueProperty('event') result = ObjectProperty(SIPResultHolding) # type: SIPResultHolding class SIPResumingEvent(SIPPluginData): sip = FixedValueProperty('event') result = ObjectProperty(SIPResultResuming) # type: SIPResultResuming class SIPHangingUpEvent(SIPPluginData): sip = FixedValueProperty('event') result = ObjectProperty(SIPResultHangingUp) # type: SIPResultHangingUp class SIPHangupEvent(SIPPluginData): sip = FixedValueProperty('event') result = ObjectProperty(SIPResultHangup) # type: SIPResultHangup class SIPIncomingCallEvent(SIPPluginData): sip = FixedValueProperty('event') result = ObjectProperty(SIPResultIncomingCall) # type: SIPResultIncomingCall call_id = StringProperty() class SIPMissedCallEvent(SIPPluginData): sip = FixedValueProperty('event') result = ObjectProperty(SIPResultMissedCall) # type: SIPResultMissedCall class SIPInfoEvent(SIPPluginData): sip = FixedValueProperty('event') result = ObjectProperty(SIPResultInfo) # type: SIPResultInfo class SIPInfoSentEvent(SIPPluginData): sip = FixedValueProperty('event') result = ObjectProperty(SIPResultInfoSent) # type: SIPResultInfoSent class SIPMessageEvent(SIPPluginData): sip = FixedValueProperty('event') result = ObjectProperty(SIPResultMessage) # type: SIPResultMessage class SIPMessageSentEvent(SIPPluginData): sip = FixedValueProperty('event') result = ObjectProperty(SIPResultMessageSent) # type: SIPResultMessageSent class SIPDTMFSentEvent(SIPPluginData): sip = FixedValueProperty('event') result = ObjectProperty(SIPResultDTMFSent) # type: SIPResultDTMFSent # Videoroom plugin data messages class VideoroomCreated(VideoroomPluginData): videoroom = FixedValueProperty('created') room = IntegerProperty() # permanent = BooleanProperty() # this is not available in older janus versions. not used. class VideoroomEdited(VideoroomPluginData): videoroom = FixedValueProperty('edited') room = IntegerProperty() # permanent = BooleanProperty() # this is not available in older janus versions. not used. class VideoroomDestroyed(VideoroomPluginData): # this comes both in the response to 'destroy' and in an event to participants still in the room when destroyed (if any) videoroom = FixedValueProperty('destroyed') room = IntegerProperty() # permanent = BooleanProperty(optional=True) # this is not available in older janus versions (only present in the response, but not in the event). not used. class VideoroomJoined(VideoroomPluginData): videoroom = FixedValueProperty('joined') room = IntegerProperty() description = StringProperty() id = IntegerProperty() publishers = ArrayProperty(VideoroomPublishers) # type: VideoroomPublishers # private_id = IntegerProperty() # this is not available in older janus versions. not used. class VideoroomAttached(VideoroomPluginData): videoroom = FixedValueProperty('attached') room = IntegerProperty() id = IntegerProperty() display = StringProperty(optional=True) class VideoroomSlowLink(VideoroomPluginData): videoroom = FixedValueProperty('slow_link') # current_bitrate = IntegerProperty() # this is actually defined as 'current-bitrate' in JSON, so we cannot map it to an attribute name. also not used. class VideoroomErrorEvent(VideoroomPluginData): __event__ = 'error' videoroom = FixedValueProperty('event') error_code = IntegerProperty() error = StringProperty() class VideoroomStartedEvent(VideoroomPluginData): __event__ = 'started' videoroom = FixedValueProperty('event') room = IntegerProperty() started = FixedValueProperty('ok') class VideoroomPublishersEvent(VideoroomPluginData): __event__ = 'publishers' videoroom = FixedValueProperty('event') room = IntegerProperty() publishers = ArrayProperty(VideoroomPublishers) # type: VideoroomPublishers class VideoroomConfiguredEvent(VideoroomPluginData): __event__ = 'configured' videoroom = FixedValueProperty('event') room = IntegerProperty() configured = FixedValueProperty('ok') class VideoroomLeftEvent(VideoroomPluginData): __event__ = 'left' videoroom = FixedValueProperty('event') room = IntegerProperty() left = FixedValueProperty('ok') class VideoroomLeavingEvent(VideoroomPluginData): __event__ = 'leaving' videoroom = FixedValueProperty('event') room = IntegerProperty() leaving = AbstractProperty() # this is either a participant id or the string "ok" reason = StringProperty(optional=True) class VideoroomKickedEvent(VideoroomPluginData): __event__ = 'kicked' videoroom = FixedValueProperty('event') room = IntegerProperty() kicked = IntegerProperty() class VideoroomUnpublishedEvent(VideoroomPluginData): __event__ = 'unpublished' videoroom = FixedValueProperty('event') room = IntegerProperty() unpublished = AbstractProperty() # this is either a participant id or the string "ok" class VideoroomPausedEvent(VideoroomPluginData): __event__ = 'paused' videoroom = FixedValueProperty('event') room = IntegerProperty() paused = FixedValueProperty('ok') class VideoroomSwitchedEvent(VideoroomPluginData): __event__ = 'switched' videoroom = FixedValueProperty('event') room = IntegerProperty() id = IntegerProperty() switched = FixedValueProperty('ok') class VideoroomJoiningEvent(VideoroomPluginData): # only sent if room has notify_joining == True (default is False). Can be used to monitor non-publishers. __event__ = 'joining' videoroom = FixedValueProperty('event') room = IntegerProperty() joining = ObjectProperty(UserId) # type: UserId class VideoroomDisplayEvent(VideoroomPluginData): # participant display name change __event__ = 'display' videoroom = FixedValueProperty('event') id = IntegerProperty() display = StringProperty() class VideoroomSubstreamEvent(VideoroomPluginData): # simulcast substream change __event__ = 'substream' videoroom = FixedValueProperty('event') room = IntegerProperty() substream = IntegerProperty() class VideoroomTemporalEvent(VideoroomPluginData): # simulcast temporal layer change __event__ = 'temporal' videoroom = FixedValueProperty('event') room = IntegerProperty() temporal = IntegerProperty() class VideoroomSpatialLayerEvent(VideoroomPluginData): # SVC spatial layer change __event__ = 'spatial_layer' videoroom = FixedValueProperty('event') room = IntegerProperty() spatial_layer = IntegerProperty() class VideoroomTemporalLayerEvent(VideoroomPluginData): # SVC temporal layer change __event__ = 'temporal_layer' videoroom = FixedValueProperty('event') room = IntegerProperty() temporal_layer = IntegerProperty() # Janus message to model mapping class ProtocolError(Exception): pass class PluginDataHandler(object): pass class SIPDataHandler(PluginDataHandler): __plugin__ = 'sip' __classmap__ = {cls.__id__: cls for cls in subclasses(SIPPluginData) if cls.__plugin__ in cls.__properties__} @classmethod def decode(cls, data): try: sip = data[cls.__plugin__] if 'error' in data: data_key = cls.__plugin__, sip, 'error' elif 'result' in data: data_key = cls.__plugin__, sip, data['result']['event'] else: data_key = cls.__plugin__, sip except KeyError: raise ProtocolError('could not get {!s} plugin data type from {!r}'.format(cls.__plugin__, data)) try: data_class = cls.__classmap__[data_key] except KeyError: raise ProtocolError('unknown {!s} plugin data: {!r}'.format(cls.__plugin__, data)) return data_class(**data) class VideoroomDataHandler(PluginDataHandler): __plugin__ = 'videoroom' __classmap__ = {cls.__id__: cls for cls in subclasses(VideoroomPluginData) if cls.__plugin__ in cls.__properties__} __eventset__ = frozenset(cls.__event__ for cls in subclasses(VideoroomPluginData) if cls.__event__) @classmethod def decode(cls, data): try: data_type = data[cls.__plugin__] except KeyError: raise ProtocolError('could not get {!s} plugin data type from {!r}'.format(cls.__plugin__, data)) if data_type == 'event': common_keys = list(cls.__eventset__.intersection(data)) if len(common_keys) != 1: raise ProtocolError('unknown {!s} plugin event: {!r}'.format(cls.__plugin__, data)) event_type = common_keys[0] data_key = cls.__plugin__, data_type, event_type else: data_key = cls.__plugin__, data_type try: data_class = cls.__classmap__[data_key] except KeyError: raise ProtocolError('unknown {!s} plugin data: {!r}'.format(cls.__plugin__, data)) return data_class(**data) class JanusMessage(object): __core_classmap__ = {cls.janus.value: cls for cls in subclasses(JanusCoreMessageBase) if 'janus' in cls.__properties__} __plugin_classmap__ = {cls.janus.value: cls for cls in subclasses(JanusPluginMessageBase) if 'janus' in cls.__properties__} __plugin_handlers__ = {handler.__plugin__: handler for handler in subclasses(PluginDataHandler)} @classmethod def from_payload(cls, payload): try: message_type = payload['janus'] except KeyError: raise ProtocolError('could not get Janus message type') if 'plugindata' in payload: try: message_class = cls.__plugin_classmap__[message_type] except KeyError: raise ProtocolError('unknown Janus message: {!s}'.format(message_type)) data_container = payload['plugindata'] try: plugin_name = data_container['plugin'].rpartition('.')[2] plugin_data = data_container['data'] except KeyError: raise ProtocolError('invalid plugin data: {!r}'.format(data_container)) try: data_handler = cls.__plugin_handlers__[plugin_name] except KeyError: raise ProtocolError('could not find a data handler for {!r}'.format(plugin_data)) data_container['data'] = data_handler.decode(plugin_data) else: try: message_class = cls.__core_classmap__[message_type] except KeyError: raise ProtocolError('unknown Janus message: {!s}'.format(message_type)) return message_class(**payload) diff --git a/sylk/applications/webrtcgateway/models/sylkpush.py b/sylk/applications/webrtcgateway/models/sylkpush.py index 4ea1699..d8665ac 100644 --- a/sylk/applications/webrtcgateway/models/sylkpush.py +++ b/sylk/applications/webrtcgateway/models/sylkpush.py @@ -1,42 +1,42 @@ -from .jsonobjects import IntegerProperty, StringProperty, FixedValueProperty +from .jsonobjects import IntegerProperty, StringProperty, FixedValueProperty, LimitedChoiceProperty from .jsonobjects import JSONObject # Event base classes (abstract, should not be used directly) class SylkRTCEventBase(JSONObject): platform = StringProperty(optional=True) app_id = StringProperty(optional=True) token = StringProperty() device_id = StringProperty(optional=True) - media_type = FixedValueProperty('video') silent = IntegerProperty(optional=True, default=0) call_id = StringProperty() def __init__(self, **kw): super(SylkRTCEventBase, self).__init__(**kw) class CallEventBase(SylkRTCEventBase): event = None # specified by subclass originator = StringProperty() from_display_name = StringProperty(optional=True, default=None) + media_type = LimitedChoiceProperty(['audio', 'video']) # Events to use used in a SylkPushRequest class ConferenceInviteEvent(CallEventBase): event = FixedValueProperty('incoming_conference_request') to = StringProperty() @property def __data__(self): data = super(ConferenceInviteEvent, self).__data__ for key in data: # Fixup keys data[key.replace('_', '-')] = data.pop(key) data['from'] = data.pop('originator') return data diff --git a/sylk/applications/webrtcgateway/models/sylkrtc.py b/sylk/applications/webrtcgateway/models/sylkrtc.py index 63a9f4e..1c308e2 100644 --- a/sylk/applications/webrtcgateway/models/sylkrtc.py +++ b/sylk/applications/webrtcgateway/models/sylkrtc.py @@ -1,436 +1,440 @@ from application.python import subclasses from .jsonobjects import BooleanProperty, IntegerProperty, StringProperty, ArrayProperty, ObjectProperty, FixedValueProperty, LimitedChoiceProperty from .jsonobjects import JSONObject, JSONArray, StringArray, CompositeValidator from .validators import AORValidator, DisplayNameValidator, LengthValidator, UniqueItemsValidator # Base models (these are abstract and should not be used directly) class SylkRTCRequestBase(JSONObject): transaction = StringProperty() class SylkRTCResponseBase(JSONObject): transaction = StringProperty() class AccountRequestBase(SylkRTCRequestBase): account = StringProperty(validator=AORValidator()) class SessionRequestBase(SylkRTCRequestBase): session = StringProperty() class VideoroomRequestBase(SylkRTCRequestBase): session = StringProperty() class AccountEventBase(JSONObject): sylkrtc = FixedValueProperty('account-event') account = StringProperty(validator=AORValidator()) class SessionEventBase(JSONObject): sylkrtc = FixedValueProperty('session-event') session = StringProperty() class VideoroomEventBase(JSONObject): sylkrtc = FixedValueProperty('videoroom-event') session = StringProperty() class AccountRegistrationStateEvent(AccountEventBase): event = FixedValueProperty('registration-state') class SessionStateEvent(SessionEventBase): event = FixedValueProperty('state') class VideoroomSessionStateEvent(VideoroomEventBase): event = FixedValueProperty('session-state') # Miscellaneous models class SIPIdentity(JSONObject): uri = StringProperty(validator=AORValidator()) display_name = StringProperty(optional=True, validator=DisplayNameValidator()) class ICECandidate(JSONObject): candidate = StringProperty() sdpMLineIndex = IntegerProperty() sdpMid = StringProperty() class ICECandidates(JSONArray): item_type = ICECandidate class AORList(StringArray): list_validator = UniqueItemsValidator() item_validator = AORValidator() class VideoroomPublisher(JSONObject): id = StringProperty() uri = StringProperty(validator=AORValidator()) display_name = StringProperty(optional=True) class VideoroomPublishers(JSONArray): item_type = VideoroomPublisher class VideoroomActiveParticipants(StringArray): list_validator = CompositeValidator(UniqueItemsValidator(), LengthValidator(maximum=2)) class VideoroomSessionOptions(JSONObject): audio = BooleanProperty(optional=True) video = BooleanProperty(optional=True) bitrate = IntegerProperty(optional=True) class VideoroomRaisedHands(StringArray): list_validator = UniqueItemsValidator() class SharedFile(JSONObject): filename = StringProperty() filesize = IntegerProperty() uploader = ObjectProperty(SIPIdentity) # type: SIPIdentity session = StringProperty() class SharedFiles(JSONArray): item_type = SharedFile # Response models class AckResponse(SylkRTCResponseBase): sylkrtc = FixedValueProperty('ack') class ErrorResponse(SylkRTCResponseBase): sylkrtc = FixedValueProperty('error') error = StringProperty() # Connection events class ReadyEvent(JSONObject): sylkrtc = FixedValueProperty('ready-event') # Account events class AccountIncomingSessionEvent(AccountEventBase): event = FixedValueProperty('incoming-session') session = StringProperty() originator = ObjectProperty(SIPIdentity) # type: SIPIdentity sdp = StringProperty() call_id = StringProperty() class AccountMissedSessionEvent(AccountEventBase): event = FixedValueProperty('missed-session') originator = ObjectProperty(SIPIdentity) # type: SIPIdentity class AccountConferenceInviteEvent(AccountEventBase): event = FixedValueProperty('conference-invite') room = StringProperty(validator=AORValidator()) session_id = StringProperty() originator = ObjectProperty(SIPIdentity) # type: SIPIdentity class AccountRegisteringEvent(AccountRegistrationStateEvent): state = FixedValueProperty('registering') class AccountRegisteredEvent(AccountRegistrationStateEvent): state = FixedValueProperty('registered') class AccountRegistrationFailedEvent(AccountRegistrationStateEvent): state = FixedValueProperty('failed') reason = StringProperty(optional=True) # Session events class SessionProgressEvent(SessionStateEvent): state = FixedValueProperty('progress') class SessionEarlyMediaEvent(SessionStateEvent): state = FixedValueProperty('early-media') sdp = StringProperty(optional=True) call_id = StringProperty(optional=True) class SessionAcceptedEvent(SessionStateEvent): state = FixedValueProperty('accepted') sdp = StringProperty(optional=True) # missing for incoming sessions call_id = StringProperty(optional=True) class SessionEstablishedEvent(SessionStateEvent): state = FixedValueProperty('established') class SessionTerminatedEvent(SessionStateEvent): state = FixedValueProperty('terminated') reason = StringProperty(optional=True) # Video room events class VideoroomConfigureEvent(VideoroomEventBase): event = FixedValueProperty('configure') originator = StringProperty() active_participants = ArrayProperty(VideoroomActiveParticipants) # type: VideoroomActiveParticipants class VideoroomSessionProgressEvent(VideoroomSessionStateEvent): state = FixedValueProperty('progress') class VideoroomSessionAcceptedEvent(VideoroomSessionStateEvent): state = FixedValueProperty('accepted') sdp = StringProperty() + video = BooleanProperty(optional=True, default=True) + audio = BooleanProperty(optional=True, default=True) class VideoroomSessionEstablishedEvent(VideoroomSessionStateEvent): state = FixedValueProperty('established') class VideoroomSessionTerminatedEvent(VideoroomSessionStateEvent): state = FixedValueProperty('terminated') reason = StringProperty(optional=True) class VideoroomFeedAttachedEvent(VideoroomEventBase): event = FixedValueProperty('feed-attached') feed = StringProperty() sdp = StringProperty() class VideoroomFeedEstablishedEvent(VideoroomEventBase): event = FixedValueProperty('feed-established') feed = StringProperty() class VideoroomInitialPublishersEvent(VideoroomEventBase): event = FixedValueProperty('initial-publishers') publishers = ArrayProperty(VideoroomPublishers) # type: VideoroomPublishers class VideoroomPublishersJoinedEvent(VideoroomEventBase): event = FixedValueProperty('publishers-joined') publishers = ArrayProperty(VideoroomPublishers) # type: VideoroomPublishers class VideoroomPublishersLeftEvent(VideoroomEventBase): event = FixedValueProperty('publishers-left') publishers = ArrayProperty(StringArray) # type: StringArray class VideoroomFileSharingEvent(VideoroomEventBase): event = FixedValueProperty('file-sharing') files = ArrayProperty(SharedFiles) # type: SharedFiles class VideoroomMessageEvent(VideoroomEventBase): event = FixedValueProperty('message') type = LimitedChoiceProperty(['normal', 'status']) content = StringProperty() content_type = StringProperty() sender = ObjectProperty(SIPIdentity) # type: SIPIdentity timestamp = StringProperty() class VideoroomComposingIndicationEvent(VideoroomEventBase): event = FixedValueProperty('composing-indication') state = StringProperty() refresh = IntegerProperty() content_type = StringProperty() sender = ObjectProperty(SIPIdentity) # type: SIPIdentity class VideoroomMessageDeliveryEvent(VideoroomEventBase): event = FixedValueProperty('message-delivery') message_id = StringProperty() delivered = BooleanProperty() code = IntegerProperty() reason = StringProperty() class VideoroomMuteAudioEvent(VideoroomEventBase): event = FixedValueProperty('mute-audio') originator = StringProperty() class VideoroomRaisedHandsEvent(VideoroomEventBase): event = FixedValueProperty('raised-hands') raised_hands = ArrayProperty(VideoroomRaisedHands) # Ping request model, can be used to check connectivity from client class PingRequest(SylkRTCRequestBase): sylkrtc = FixedValueProperty('ping') # Account request models class AccountAddRequest(AccountRequestBase): sylkrtc = FixedValueProperty('account-add') password = StringProperty(validator=LengthValidator(minimum=1, maximum=9999)) display_name = StringProperty(optional=True) user_agent = StringProperty(optional=True) class AccountRemoveRequest(AccountRequestBase): sylkrtc = FixedValueProperty('account-remove') class AccountRegisterRequest(AccountRequestBase): sylkrtc = FixedValueProperty('account-register') class AccountUnregisterRequest(AccountRequestBase): sylkrtc = FixedValueProperty('account-unregister') class AccountDeviceTokenRequest(AccountRequestBase): sylkrtc = FixedValueProperty('account-devicetoken') token = StringProperty() platform = StringProperty() device = StringProperty() silent = BooleanProperty(default=False) app = StringProperty() # Session request models class SessionCreateRequest(SessionRequestBase): sylkrtc = FixedValueProperty('session-create') account = StringProperty(validator=AORValidator()) uri = StringProperty(validator=AORValidator()) sdp = StringProperty() class SessionAnswerRequest(SessionRequestBase): sylkrtc = FixedValueProperty('session-answer') sdp = StringProperty() class SessionTrickleRequest(SessionRequestBase): sylkrtc = FixedValueProperty('session-trickle') candidates = ArrayProperty(ICECandidates) # type: ICECandidates class SessionTerminateRequest(SessionRequestBase): sylkrtc = FixedValueProperty('session-terminate') # Videoroom request models class VideoroomJoinRequest(VideoroomRequestBase): sylkrtc = FixedValueProperty('videoroom-join') account = StringProperty(validator=AORValidator()) uri = StringProperty(validator=AORValidator()) sdp = StringProperty() + audio = BooleanProperty(optional=True, default=True) + video = BooleanProperty(optional=True, default=True) class VideoroomLeaveRequest(VideoroomRequestBase): sylkrtc = FixedValueProperty('videoroom-leave') class VideoroomConfigureRequest(VideoroomRequestBase): sylkrtc = FixedValueProperty('videoroom-configure') active_participants = ArrayProperty(VideoroomActiveParticipants) # type: VideoroomActiveParticipants class VideoroomFeedAttachRequest(VideoroomRequestBase): sylkrtc = FixedValueProperty('videoroom-feed-attach') publisher = StringProperty() feed = StringProperty() class VideoroomFeedAnswerRequest(VideoroomRequestBase): sylkrtc = FixedValueProperty('videoroom-feed-answer') feed = StringProperty() sdp = StringProperty() class VideoroomFeedDetachRequest(VideoroomRequestBase): sylkrtc = FixedValueProperty('videoroom-feed-detach') feed = StringProperty() class VideoroomInviteRequest(VideoroomRequestBase): sylkrtc = FixedValueProperty('videoroom-invite') participants = ArrayProperty(AORList) # type: AORList class VideoroomSessionTrickleRequest(VideoroomRequestBase): sylkrtc = FixedValueProperty('videoroom-session-trickle') candidates = ArrayProperty(ICECandidates) # type: ICECandidates class VideoroomSessionUpdateRequest(VideoroomRequestBase): sylkrtc = FixedValueProperty('videoroom-session-update') options = ObjectProperty(VideoroomSessionOptions) # type: VideoroomSessionOptions class VideoroomMessageRequest(VideoroomRequestBase): sylkrtc = FixedValueProperty('videoroom-message') message_id = StringProperty() content = StringProperty() content_type = StringProperty() class VideoroomComposingIndicationRequest(VideoroomRequestBase): sylkrtc = FixedValueProperty('videoroom-composing-indication') state = LimitedChoiceProperty(['active', 'idle']) refresh = IntegerProperty(optional=True) class VideoroomMuteAudioParticipantsRequest(VideoroomRequestBase): sylkrtc = FixedValueProperty('videoroom-mute-audio-participants') class VideoroomToggleHandRequest(VideoroomRequestBase): sylkrtc = FixedValueProperty('videoroom-toggle-hand') session_id = StringProperty(optional=True) # SylkRTC request to model mapping class ProtocolError(Exception): pass class SylkRTCRequest(object): __classmap__ = {cls.sylkrtc.value: cls for cls in subclasses(SylkRTCRequestBase) if hasattr(cls, 'sylkrtc')} @classmethod def from_message(cls, message): try: request_type = message['sylkrtc'] except KeyError: raise ProtocolError('could not get WebSocket message type') try: request_class = cls.__classmap__[request_type] except KeyError: raise ProtocolError('unknown WebSocket request: %s' % request_type) return request_class(**message) diff --git a/sylk/applications/webrtcgateway/push.py b/sylk/applications/webrtcgateway/push.py index fa43140..b426b8b 100644 --- a/sylk/applications/webrtcgateway/push.py +++ b/sylk/applications/webrtcgateway/push.py @@ -1,87 +1,93 @@ import json from twisted.internet import defer, reactor from twisted.web.client import Agent, readBody from twisted.web.iweb import IBodyProducer from twisted.web.http_headers import Headers from zope.interface import implementer from .configuration import GeneralConfig from .logger import log from .models import sylkpush from .storage import TokenStorage __all__ = 'conference_invite' agent = Agent(reactor) headers = Headers({'User-Agent': ['SylkServer'], 'Content-Type': ['application/json']}) @implementer(IBodyProducer) class StringProducer(object): def __init__(self, data): self.body = data self.length = len(data) def startProducing(self, consumer): consumer.write(self.body) return defer.succeed(None) def pauseProducing(self): pass def stopProducing(self): pass def _construct_and_send(result, request, destination): for device_token, push_parameters in result.iteritems(): request.token = device_token request.app_id = push_parameters['app'] request.platform = push_parameters['platform'] request.device_id = push_parameters['device_id'] _send_push_notification(request, destination) -def conference_invite(originator, destination, room, call_id): +def conference_invite(originator, destination, room, call_id, audio, video): tokens = TokenStorage() + if video: + media_type = video + else: + media_type = audio + request = sylkpush.ConferenceInviteEvent(token='dummy', app_id='dummy', platform='dummy', device_id='dummy', - originator=originator.uri, from_display_name=originator.display_name, to=room, call_id=str(call_id)) + originator=originator.uri, from_display_name=originator.display_name, to=room, call_id=str(call_id), + media_type=media_type) user_tokens = tokens[destination] if isinstance(user_tokens, set): return else: if isinstance(user_tokens, defer.Deferred): user_tokens.addCallback(lambda result: _construct_and_send(result, request, destination)) else: _construct_and_send(user_tokens, request, destination) @defer.inlineCallbacks def _send_push_notification(payload, destination): if GeneralConfig.sylk_push_url: try: r = yield agent.request('POST', GeneralConfig.sylk_push_url, headers, StringProducer(json.dumps(payload.__data__))) except Exception as e: log.info('Error sending push notification to %s: %s', GeneralConfig.sylk_push_url, e) else: if r.code != 200: if r.code == 410: try: body = yield readBody(r) except Exception as e: log.warning('Error reading response body: %s', e) else: if 'application/json' in r.headers.getRawHeaders('content-type'): token = json.loads(body)['data']['token'] log.info('Token expired, purging old token from storage') tokens = TokenStorage() tokens.remove(destination, token) else: log.warning('Error sending push notification: %s', r.phrase) else: log.info('Sent push notification: %s', payload) else: log.warning('Cannot send push notification: no Sylk push server configured')