diff --git a/sylk-server b/sylk-server index 2dfdef5..3c9ab3a 100644 --- a/sylk-server +++ b/sylk-server @@ -1,114 +1,114 @@ #!/usr/bin/env python import os import signal import sys from application import log from application.process import process, ProcessError from optparse import OptionParser import sipsimple import sylk def main(): name = 'sylk-server' fullname = 'SylkServer' runtime_directory = '/var/run/sylkserver' system_config_directory = '/etc/sylkserver' default_pid = os.path.join(runtime_directory, 'server.pid') default_config = sylk.configuration_filename if os.path.isfile(sylk.configuration_filename) else os.path.join(system_config_directory, sylk.configuration_filename) parser = OptionParser(version='%%prog %s' % sylk.__version__) parser.add_option('--no-fork', action='store_false', dest='fork', default=1, help='run the process in the foreground (for debugging)') parser.add_option('--pid', dest='pid_file', help='pid file (%r)' % default_pid, metavar='File') parser.add_option('--config-file', dest='config_file', default=default_config, help='path to configuration file to read (%r)' % default_config, metavar='File') parser.add_option('--enable-bonjour', action='store_true', dest='enable_bonjour', default=False, help='enable Bonjour services') parser.add_option('--debug-memory', action='store_true', dest='debug_memory', default=False, help='enable memory debugging (only works with --no-fork)') options, args = parser.parse_args() path, configuration_filename = os.path.split(options.config_file) if path: system_config_directory = path process.system_config_directory = system_config_directory sylk.configuration_filename = process.config_file(options.config_file) pid_file = options.pid_file or default_pid # when run in foreground, do not require root access because of /var/run/sylkserver if not options.fork: process._runtime_directory = None else: try: process.runtime_directory = runtime_directory process.daemonize(pid_file) except ProcessError as e: log.fatal('Cannot start {name}: {exception!s}'.format(name=fullname, exception=e)) sys.exit(1) log.start_syslog(name) from sylk.server import SylkServer, ServerConfig log.info('Starting {name} {module.__version__}, using SIP SIMPLE SDK {sipsimple.__version__}'.format(name=fullname, module=sylk, sipsimple=sipsimple)) config_file = ServerConfig.__cfgtype__(ServerConfig.__cfgfile__) if config_file.files: log.info('Reading configuration from {}'.format(', '.join(config_file.files))) else: log.info('Not reading any configuration files (using internal defaults)') if not options.fork and options.debug_memory: import atexit from application.debug.memory import memory_dump atexit.register(memory_dump) server = SylkServer() def stop_server(*args): if not server.stopping_event.is_set(): log.info('Stopping {name}...'.format(name=fullname)) server.stop() process.signals.add_handler(signal.SIGTERM, stop_server) process.signals.add_handler(signal.SIGINT, stop_server) def toggle_debugging(*args): from sylk.configuration import ServerConfig if log.level.current != log.level.DEBUG: log.level.current = log.level.DEBUG log.info('Switched logging level to DEBUG') else: log.info('Switched logging level to {}'.format(ServerConfig.log_level)) log.level.current = ServerConfig.log_level process.signals.add_handler(signal.SIGUSR1, toggle_debugging) try: server.start(options) except Exception as e: log.fatal('Failed to start {name}: {exception!s}'.format(name=fullname, exception=e)) log.exception() sys.exit(1) else: while not server.stopping_event.wait(9999): pass - server.stop_event.wait(5) - if server.stop_event.is_set(): + server.stopped_event.wait(5) + if server.stopped_event.is_set(): log.info('{name} stopped'.format(name=fullname)) else: log.info('Forcefully exiting {name}...'.format(name=fullname)) os._exit(1) sys.exit(int(server.failed)) if __name__ == "__main__": main() diff --git a/sylk/server.py b/sylk/server.py index 4d38f21..a30be90 100644 --- a/sylk/server.py +++ b/sylk/server.py @@ -1,247 +1,247 @@ import os import sys from threading import Event from uuid import uuid4 from application import log from application.notification import NotificationCenter from application.python import Null from application.system import makedirs from eventlib import proc from sipsimple.account import Account, BonjourAccount, AccountManager from sipsimple.application import SIPApplication from sipsimple.audio import AudioDevice, RootAudioBridge from sipsimple.configuration.settings import SIPSimpleSettings from sipsimple.core import AudioMixer from sipsimple.lookup import DNSManager from sipsimple.storage import MemoryStorage from sipsimple.threading import ThreadManager from sipsimple.threading.green import run_in_green_thread from sipsimple.video import VideoDevice from twisted.internet import reactor # Load stream extensions needed for integration with SIP SIMPLE SDK import sylk.streams; del sylk.streams from sylk.accounts import DefaultAccount from sylk.applications import IncomingRequestHandler from sylk.configuration import ServerConfig, SIPConfig, ThorNodeConfig from sylk.configuration.settings import AccountExtension, BonjourAccountExtension, SylkServerSettingsExtension from sylk.log import TraceLogManager from sylk.session import SessionManager from sylk.web import WebServer class SylkServer(SIPApplication): def __init__(self): self.request_handler = Null self.thor_interface = Null self.web_server = Null self.options = Null self.stopping_event = Event() - self.stop_event = Event() + self.stopped_event = Event() self.failed = False def start(self, options): self.options = options if self.options.enable_bonjour: ServerConfig.enable_bonjour = True notification_center = NotificationCenter() notification_center.add_observer(self, sender=self) notification_center.add_observer(self, name='ThorNetworkGotFatalError') Account.register_extension(AccountExtension) BonjourAccount.register_extension(BonjourAccountExtension) SIPSimpleSettings.register_extension(SylkServerSettingsExtension) super(SylkServer, self).start(MemoryStorage()) def _initialize_core(self): # SylkServer needs to listen for extra events and request types notification_center = NotificationCenter() settings = SIPSimpleSettings() # initialize core options = dict(# general ip_address=SIPConfig.local_ip, user_agent=settings.user_agent, # SIP detect_sip_loops=False, udp_port=settings.sip.udp_port if 'udp' in settings.sip.transport_list else None, tcp_port=settings.sip.tcp_port if 'tcp' in settings.sip.transport_list else None, tls_port=None, # TLS tls_verify_server=False, tls_ca_file=None, tls_cert_file=None, tls_privkey_file=None, # rtp rtp_port_range=(settings.rtp.port_range.start, settings.rtp.port_range.end), # audio codecs=list(settings.rtp.audio_codec_list), # video video_codecs=list(settings.rtp.video_codec_list), enable_colorbar_device=True, # logging log_level=settings.logs.pjsip_level if settings.logs.trace_pjsip else 0, trace_sip=settings.logs.trace_sip, # events and requests to handle events={'conference': ['application/conference-info+xml'], 'presence': ['application/pidf+xml'], 'refer': ['message/sipfrag;version=2.0']}, incoming_events={'conference', 'presence'}, incoming_requests={'MESSAGE'}) notification_center.add_observer(self, sender=self.engine) self.engine.start(**options) @run_in_green_thread def _initialize_subsystems(self): notification_center = NotificationCenter() with self._lock: stop_pending = self._stop_pending if stop_pending: self.state = 'stopping' if stop_pending: notification_center.post_notification('SIPApplicationWillEnd', sender=self) reactor.stop() return account_manager = AccountManager() dns_manager = DNSManager() session_manager = SessionManager() settings = SIPSimpleSettings() # Initialize default account default_account = DefaultAccount() account_manager.default_account = default_account # initialize TLS self._initialize_tls() # initialize PJSIP internal resolver self.engine.set_nameservers(dns_manager.nameservers) # initialize audio objects voice_mixer = AudioMixer(None, None, settings.audio.sample_rate, 0, 9999) self.voice_audio_device = AudioDevice(voice_mixer) self.voice_audio_bridge = RootAudioBridge(voice_mixer) self.voice_audio_bridge.add(self.voice_audio_device) # initialize video objects self.video_device = VideoDevice(u'Colorbar generator', settings.video.resolution, settings.video.framerate) # initialize instance id settings.instance_id = uuid4().urn settings.save() # initialize ZRTP cache makedirs(ServerConfig.spool_dir.normalized) self.engine.zrtp_cache = os.path.join(ServerConfig.spool_dir.normalized, 'zrtp.db') # initialize middleware components dns_manager.start() account_manager.start() session_manager.start() notification_center.add_observer(self, name='CFGSettingsObjectDidChange') self.state = 'started' notification_center.post_notification('SIPApplicationDidStart', sender=self) # start SylkServer components self.web_server = WebServer() self.web_server.start() self.request_handler = IncomingRequestHandler() self.request_handler.start() if ThorNodeConfig.enabled: from sylk.interfaces.sipthor import ConferenceNode self.thor_interface = ConferenceNode() thor_roles = [] if 'conference' in self.request_handler.application_registry: thor_roles.append('conference_server') if 'xmppgateway' in self.request_handler.application_registry: thor_roles.append('xmpp_gateway') if 'webrtcgateway' in self.request_handler.application_registry: thor_roles.append('webrtc_gateway') self.thor_interface.start(thor_roles) @run_in_green_thread def _shutdown_subsystems(self): dns_manager = DNSManager() account_manager = AccountManager() session_manager = SessionManager() # terminate all sessions p = proc.spawn(session_manager.stop) p.wait() # shutdown SylkServer components procs = [proc.spawn(self.web_server.stop), proc.spawn(self.request_handler.stop), proc.spawn(self.thor_interface.stop)] proc.waitall(procs) # shutdown other middleware components procs = [proc.spawn(dns_manager.stop), proc.spawn(account_manager.stop)] proc.waitall(procs) # shutdown engine self.engine.stop() self.engine.join(timeout=5) # stop threads thread_manager = ThreadManager() thread_manager.stop() # stop the reactor reactor.stop() def _NH_AudioDevicesDidChange(self, notification): pass def _NH_DefaultAudioDeviceDidChange(self, notification): pass def _NH_SIPApplicationFailedToStartTLS(self, notification): log.fatal('Could not set TLS options: %s' % notification.data.error) sys.exit(1) def _NH_SIPApplicationWillStart(self, notification): tracelog_manager = TraceLogManager() tracelog_manager.start() def _NH_SIPApplicationDidStart(self, notification): settings = SIPSimpleSettings() local_ip = SIPConfig.local_ip log.info('SylkServer started, listening on:') for transport in settings.sip.transport_list: try: log.info(' %s:%d (%s)' % (local_ip, getattr(self.engine, '%s_port' % transport), transport.upper())) except TypeError: pass def _NH_SIPApplicationWillEnd(self, notification): self.stopping_event.set() def _NH_SIPApplicationDidEnd(self, notification): log.info('SIP application ended') tracelog_manager = TraceLogManager() tracelog_manager.stop() if not self.stopping_event.is_set(): log.warning('SIP application ended without shutting down all subsystems') self.stopping_event.set() - self.stop_event.set() + self.stopped_event.set() def _NH_SIPEngineDidFail(self, notification): log.error('SIP engine failed') self.failed = True super(SylkServer, self)._NH_SIPEngineDidFail(notification) def _NH_ThorNetworkGotFatalError(self, notification): log.error('All Thor Event Servers have unrecoverable errors')