<commit_before>_u = require("underscore")
url = require('url')
http = require "http"
express = require "express"
Logger = require "./log_controller"

module.exports = class Core
    DefaultOptions:
        streams:            {}
        # after a new deployment, allow a one hour grace period for 
        # connected listeners
        max_zombie_life:    1000 * 60 * 60
        
    Redis:  require "./redis_config"
    Stream: require "./stream"
    Rewind: require "./rewind_buffer"
    Preroller: require "./preroller"
    Server: require "./server"
        
    Sources:
        proxy:  require("./sources/proxy_room")
        
    Outputs:
        pumper:     require("./outputs/pumper")
        shoutcast:  require("./outputs/shoutcast")
        mp3:        require("./outputs/livemp3")
        sockets:    require("./outputs/sockets")
        
    #----------
        
    constructor: ->        
        @streams = {}
        
        @root_route = null
        
        # -- Don't crash on uncaughtException errors -- #
        
        process.on "uncaughtException", (error) =>
          @log.error "uncaught exception: " + error, stack:error.stack        

        # -- Graceful Shutdown -- #
        
        # attach a USR2 handler that causes us to stop listening. this is 
        # used to allow a new server to start up
        process.on "SIGTERM", =>
            @log.debug "Got SIGTERM in core...  Starting shutdown"
            
            # stop accepting new connections
            @server?.stopListening()
            
            # when should we get pushy?
            @_shutdownMaxTime = (new Date).getTime() + @options.max_zombie_life
            
            # need to start a process to quit when existing connections disconnect
            @_shutdownTimeout = setInterval =>
                # have we used up the grace period yet?
                force_shut = if (new Date).getTime() > @_shutdownMaxTime then true else false
                
                # build a listener count for all streams
                listeners = 0
                for k,stream of @streams
                    slisteners = stream.countListeners()
                    
                    if slisteners == 0 || force_shut
                        # if everyone's done (or we're feeling pushy), disconnect
                        stream.disconnect()
                    else
                        listeners += slisteners
                    
                if listeners == 0
                    # everyone's out...  go ahead and shut down
                    @log.debug "Shutdown complete", => process.exit()
                else
                    @log.debug "Still awaiting shutdown; #{listeners} listeners"
                
            , 60 * 1000
                
    #----------
        
    # configureStreams can be called on a new core, or it can be called to 
    # reconfigure an existing core.  we need to support either one.
    configureStreams: (options) ->
        @log.debug "In configure with ", options:options

        # -- Sources -- #
        
        # are any of our current streams missing from the new options? if so, 
        # disconnect them
        for k,obj in @streams
            console.log "calling disconnect on ", k
            obj.disconnect() unless options.streams?[k]
            @streams.delete k
        
        # run through the streams we've been passed, initializing sources and 
        # creating rewind buffers
        for key,opts of options.streams
            console.log "stream for #{key}"
            if @streams[key]
                # existing stream...  pass it updated configuration
                @log.debug "Passing updated config to source: #{key}", opts:opts
                @streams[key].configure opts
            else
                @log.debug "Starting up source: #{key}", opts:opts
                @streams[key] = new @Stream @, key, @log.child(stream:key), opts
                
            # should this stream accept requests to /?
            if opts.root_route
                @root_route = key
                        
    #----------
    
    class @Standalone extends Core
        constructor: (opts) ->
            @options = _u.defaults opts||{}, @DefaultOptions
            
            # -- Set up logging -- #
            
            @log = new Logger @options.log
            @log.debug("Instance initialized")
            
            # -- run Core's constructor -- #
            
            super()
            
            # -- set up our stream server -- #
            
            # init our server
            @server = new @Server @
            @server.listen @options.listen
                
            @log.debug "Standalone is listening on port #{@options.listen}"
        
            # start up the socket manager on the listener
            @sockets = new @Outputs.sockets server:@server.server, core:@
            
            # -- initialize streams -- #
            
            @configureStreams streams:@options.streams
    
    #----------
    
    # Master Server
    # 
    # Masters don't handle stream connections directly.  Instead they host 
    # configuration info and pass it on to slave servers.  They also consolidate 
    # slave logging data.
    
    class @Master extends Core
        constructor: (opts) ->
            @options = _u.defaults opts||{}, @DefaultOptions
            
            @slaves = []
            
            # -- set up logging -- #
            
            @log = new Logger @options.log
            @log.debug("Instance initialized")
            
            # -- allow Core's constructor to run -- #
            
            super()
            
            # -- load our streams configuration from redis -- #
            
            @redis = new @Redis @options.redis
            @redis.on "config", (streams) =>
                # stash the configuration
                @options.streams = streams
                
                # and then pass it on to any connected slaves
                for sock in @slaves
                    sock.emit "config", streams:streams
            
            # -- set up the socket connection for slaves -- #
            
            if !@options.master?.port || !@options.master?.password
                @log.error("Invalid options for master server.  Must provide port and password.")
                return false                
                
            # fire up a socket listener on our slave port
            @io = require("socket.io").listen @options.master?.port
            
            # by calling this @server, it will get disconnected on SIGTERM
            @server = @io
            
            # add our authentication<commit_msg>Fix master handler for stopping listening on TERM<commit_after>_u = require("underscore")
url = require('url')
http = require "http"
express = require "express"
Logger = require "./log_controller"

module.exports = class Core
    DefaultOptions:
        streams:            {}
        # after a new deployment, allow a one hour grace period for 
        # connected listeners
        max_zombie_life:    1000 * 60 * 60
        
    Redis:  require "./redis_config"
    Stream: require "./stream"
    Rewind: require "./rewind_buffer"
    Preroller: require "./preroller"
    Server: require "./server"
        
    Sources:
        proxy:  require("./sources/proxy_room")
        
    Outputs:
        pumper:     require("./outputs/pumper")
        shoutcast:  require("./outputs/shoutcast")
        mp3:        require("./outputs/livemp3")
        sockets:    require("./outputs/sockets")
        
    #----------
        
    constructor: ->        
        @streams = {}
        
        @root_route = null
        
        # -- Don't crash on uncaughtException errors -- #
        
        process.on "uncaughtException", (error) =>
          @log.error "uncaught exception: " + error, stack:error.stack        

        # -- Graceful Shutdown -- #
        
        # attach a USR2 handler that causes us to stop listening. this is 
        # used to allow a new server to start up
        process.on "SIGTERM", =>
            @log.debug "Got SIGTERM in core...  Starting shutdown"
            
            # stop accepting new connections
            @server?.stopListening()
            
            # when should we get pushy?
            @_shutdownMaxTime = (new Date).getTime() + @options.max_zombie_life
            
            # need to start a process to quit when existing connections disconnect
            @_shutdownTimeout = setInterval =>
                # have we used up the grace period yet?
                force_shut = if (new Date).getTime() > @_shutdownMaxTime then true else false
                
                # build a listener count for all streams
                listeners = 0
                for k,stream of @streams
                    slisteners = stream.countListeners()
                    
                    if slisteners == 0 || force_shut
                        # if everyone's done (or we're feeling pushy), disconnect
                        stream.disconnect()
                    else
                        listeners += slisteners
                    
                if listeners == 0
                    # everyone's out...  go ahead and shut down
                    @log.debug "Shutdown complete", => process.exit()
                else
                    @log.debug "Still awaiting shutdown; #{listeners} listeners"
                
            , 60 * 1000
                
    #----------
        
    # configureStreams can be called on a new core, or it can be called to 
    # reconfigure an existing core.  we need to support either one.
    configureStreams: (options) ->
        @log.debug "In configure with ", options:options

        # -- Sources -- #
        
        # are any of our current streams missing from the new options? if so, 
        # disconnect them
        for k,obj in @streams
            console.log "calling disconnect on ", k
            obj.disconnect() unless options.streams?[k]
            @streams.delete k
        
        # run through the streams we've been passed, initializing sources and 
        # creating rewind buffers
        for key,opts of options.streams
            console.log "stream for #{key}"
            if @streams[key]
                # existing stream...  pass it updated configuration
                @log.debug "Passing updated config to source: #{key}", opts:opts
                @streams[key].configure opts
            else
                @log.debug "Starting up source: #{key}", opts:opts
                @streams[key] = new @Stream @, key, @log.child(stream:key), opts
                
            # should this stream accept requests to /?
            if opts.root_route
                @root_route = key
                        
    #----------
    
    class @Standalone extends Core
        constructor: (opts) ->
            @options = _u.defaults opts||{}, @DefaultOptions
            
            # -- Set up logging -- #
            
            @log = new Logger @options.log
            @log.debug("Instance initialized")
            
            # -- run Core's constructor -- #
            
            super()
            
            # -- set up our stream server -- #
            
            # init our server
            @server = new @Server @
            @server.listen @options.listen
                
            @log.debug "Standalone is listening on port #{@options.listen}"
        
            # start up the socket manager on the listener
            @sockets = new @Outputs.sockets server:@server.server, core:@
            
            # -- initialize streams -- #
            
            @configureStreams streams:@options.streams
    
    #----------
    
    # Master Server
    # 
    # Masters don't handle stream connections directly.  Instead they host 
    # configuration info and pass it on to slave servers.  They also consolidate 
    # slave logging data.
    
    class @Master extends Core
        constructor: (opts) ->
            @options = _u.defaults opts||{}, @DefaultOptions
            
            @slaves = []
            
            # -- set up logging -- #
            
            @log = new Logger @options.log
            @log.debug("Instance initialized")
            
            # -- allow Core's constructor to run -- #
            
            super()
            
            # -- load our streams configuration from redis -- #
            
            @redis = new @Redis @options.redis
            @redis.on "config", (streams) =>
                # stash the configuration
                @options.streams = streams
                
                # and then pass it on to any connected slaves
                for sock in @slaves
                    sock.emit "config", streams:streams
            
            # -- set up the socket connection for slaves -- #
            
            if !@options.master?.port || !@options.master?.password
                @log.error("Invalid options for master server.  Must provide port and password.")
                return false                
                
            # fire up a socket listener on our slave port
            @io = require("socket.io").listen @options.master?.port
            
            # disconnect from our port on SIGTERM
            process.on "SIGTERM", => @io.close()
            
            # add our authentication
{'fraction_non_alphanumeric': 0.0896128841607565, 'fraction_numerical': 0.002955082742316785, 'mean_word_length': 1.294017963057109, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 24, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 4, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '7633824', 'n_tokens_mistral': 3329, 'n_tokens_neox': 3167, 'n_words': 1057}