<commit_before>_u = require "underscore"
EventEmitter = require('events').EventEmitter

module.exports = class Stream extends EventEmitter
    DefaultOptions:
        meta_interval:  32768
        max_buffer:     4194304 # 4 megabits (64 seconds of 64k audio)
        name:           ""
        
    constructor: (@core,key,log,opts) ->
        @options = _u.defaults opts||{}, @DefaultOptions
        
        # remove our max listener count
        @setMaxListeners 0
            
        @log = log
        @key = key
        
        @_id_increment = 1
        @_lmeta = {}
        
        @preroll = null
        @mlog_timer = null

        @dataFunc = (chunk) => (l.data(chunk) if l.data) for id,l of @_lmeta                    
        @metaFunc = (chunk) => (l.meta(chunk) if l.meta) for id,l of @_lmeta
        
        # now run configure...
        process.nextTick => @configure(opts)
                                    
        # set up a rewind buffer
        @rewind = new @core.Rewind @, opts.rewind
            
    #----------
        
    configure: (opts) ->
        # -- Source -- #
        
        # make sure our source is valid
        if @core.Sources[ opts.source?.type ]
            # stash our existing source if we have one
            old_source = @source || null
            
            # bring the new / only source up
            newsource = new @core.Sources[ opts.source.type ] @, @key, opts.source
            process.nextTick =>
                if newsource.connect()
                    # look for a header to say the new source is up and running
                    newsource.once "header", =>
                        
                        if old_source
                            # unhook from the old source's events
                            old_source.removeListener "metadata",   @metaFunc
                            old_source.removeListener "data",       @dataFunc
                    
                        @source = newsource
                    
                        # connect to the new source's events
                        newsource.on "metadata",   @metaFunc
                        newsource.on "data",       @dataFunc
                
                        # note that we've got a new source
                        process.nextTick =>
                            console.log "emit source event"
                            @emit "source", newsource

                        # disconnect the old source, which we're now no longer using
                        old_source?.disconnect()
                else
                    @log.error "Failed to connect to new source"
        else
            @log.error "Invalid source type.", opts:opts
            return false
                        
        # -- Preroll -- #
        
        @log.debug "Preroll settings are ", preroll:opts.preroll
        
        @preroll.disconnect() if @preroll
        
        if opts.preroll?
            # create a Preroller connection
            @preroll = new @core.Preroller @, @key, opts.preroll
        
        # -- Should we be logging minutes? -- #
        
        if opts.log_minutes && !@mlog_timer
            # set up a timer to log minutes listened
            @mlog_timer = setInterval => 
                # run through each listener and log minute or time listened
                now = new Date
                for id,l of @_lmeta
                    # we only log requests that have lasted more than one minute. 
                    # we use startTime for the length the session has been going, 
                    # but use minuteTime for accumulating unlogged time 
                    if ( now.getTime() / 1000 - l.startTime.getTime() / 1000 ) > 60
                        dur = ( now.getTime() / 1000 - l.minuteTime.getTime() / 1000 )
                        @log.minute "", path:l.obj.reqPath, time:now, ua:l.obj.reqUA, duration:dur
                        l.minuteTime = now
                                        
            , 60*1000
            
        else if @mlog_timer && !opts.log_minutes
            clearInterval @mlog_timer
            @mlog_timer = null
            
        # -- Set up bufferSize poller -- #
        
        # We disconnect clients that have fallen too far behind on their 
        # buffers. Buffer size can be configured via the "max_buffer" setting, 
        # which takes bits
        console.log "max buffer size is ", @options.max_buffer
        @buf_timer = setInterval =>
            all_buf = 0
            for id,l of @_lmeta
                conn = l.obj.req?.socket
                
                all_buf += conn?.bufferSize||0
                
                if (conn?.bufferSize||0) > @options.max_buffer
                    @log.debug "Connection exceeded max buffer size.", req:l.obj.req, bufferSize:conn.bufferSize
                    l.obj.disconnect()

            @log.debug "All buffers: #{all_buf}"
        , 60*1000
                
                        
    #----------
        
    disconnect: ->
        # handle clearing out lmeta
        l.obj.disconnect(true) for k,l of @_lmeta
            
        # disconnect the stream source
        @source?.disconnect()
        
    #----------
    <commit_msg>Add force flag to max buffer disconnect call; fixes memory leak<commit_after>_u = require "underscore"
EventEmitter = require('events').EventEmitter

module.exports = class Stream extends EventEmitter
    DefaultOptions:
        meta_interval:  32768
        max_buffer:     4194304 # 4 megabits (64 seconds of 64k audio)
        name:           ""
        
    constructor: (@core,key,log,opts) ->
        @options = _u.defaults opts||{}, @DefaultOptions
        
        # remove our max listener count
        @setMaxListeners 0
            
        @log = log
        @key = key
        
        @_id_increment = 1
        @_lmeta = {}
        
        @preroll = null
        @mlog_timer = null

        @dataFunc = (chunk) => (l.data(chunk) if l.data) for id,l of @_lmeta                    
        @metaFunc = (chunk) => (l.meta(chunk) if l.meta) for id,l of @_lmeta
        
        # now run configure...
        process.nextTick => @configure(opts)
                                    
        # set up a rewind buffer
        @rewind = new @core.Rewind @, opts.rewind
            
    #----------
        
    configure: (opts) ->
        # -- Source -- #
        
        # make sure our source is valid
        if @core.Sources[ opts.source?.type ]
            # stash our existing source if we have one
            old_source = @source || null
            
            # bring the new / only source up
            newsource = new @core.Sources[ opts.source.type ] @, @key, opts.source
            process.nextTick =>
                if newsource.connect()
                    # look for a header to say the new source is up and running
                    newsource.once "header", =>
                        
                        if old_source
                            # unhook from the old source's events
                            old_source.removeListener "metadata",   @metaFunc
                            old_source.removeListener "data",       @dataFunc
                    
                        @source = newsource
                    
                        # connect to the new source's events
                        newsource.on "metadata",   @metaFunc
                        newsource.on "data",       @dataFunc
                
                        # note that we've got a new source
                        process.nextTick =>
                            console.log "emit source event"
                            @emit "source", newsource

                        # disconnect the old source, which we're now no longer using
                        old_source?.disconnect()
                else
                    @log.error "Failed to connect to new source"
        else
            @log.error "Invalid source type.", opts:opts
            return false
                        
        # -- Preroll -- #
        
        @log.debug "Preroll settings are ", preroll:opts.preroll
        
        @preroll.disconnect() if @preroll
        
        if opts.preroll?
            # create a Preroller connection
            @preroll = new @core.Preroller @, @key, opts.preroll
        
        # -- Should we be logging minutes? -- #
        
        if opts.log_minutes && !@mlog_timer
            # set up a timer to log minutes listened
            @mlog_timer = setInterval => 
                # run through each listener and log minute or time listened
                now = new Date
                for id,l of @_lmeta
                    # we only log requests that have lasted more than one minute. 
                    # we use startTime for the length the session has been going, 
                    # but use minuteTime for accumulating unlogged time 
                    if ( now.getTime() / 1000 - l.startTime.getTime() / 1000 ) > 60
                        dur = ( now.getTime() / 1000 - l.minuteTime.getTime() / 1000 )
                        @log.minute "", path:l.obj.reqPath, time:now, ua:l.obj.reqUA, duration:dur
                        l.minuteTime = now
                                        
            , 60*1000
            
        else if @mlog_timer && !opts.log_minutes
            clearInterval @mlog_timer
            @mlog_timer = null
            
        # -- Set up bufferSize poller -- #
        
        # We disconnect clients that have fallen too far behind on their 
        # buffers. Buffer size can be configured via the "max_buffer" setting, 
        # which takes bits
        console.log "max buffer size is ", @options.max_buffer
        @buf_timer = setInterval =>
            all_buf = 0
            for id,l of @_lmeta
                conn = l.obj.req?.socket
                
                all_buf += conn?.bufferSize||0
                
                if (conn?.bufferSize||0) > @options.max_buffer
                    @log.debug "Connection exceeded max buffer size.", req:l.obj.req, bufferSize:conn.bufferSize
                    l.obj.disconnect(true)

            @log.debug "All buffers: #{all_buf}"
        , 60*1000
                
                        
    #----------
        
    disconnect: ->
        # handle clearing out lmeta
        l.obj.disconnect(true) for k,l of @_lmeta
            
        # disconnect the stream source
        @source?.disconnect()
        
    #----------
    
{'fraction_non_alphanumeric': 0.08358576296857251, 'fraction_numerical': 0.009844755774327906, 'mean_word_length': 1.1451776649746193, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 29, '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': '4988539', 'n_tokens_mistral': 2770, 'n_tokens_neox': 2545, 'n_words': 792}