<commit_before>_       = require "underscore"
http    = require "http"
url     = require "url"
request = require "request"
xmldom  = require "xmldom"
xpath   = require "xpath"

debug = require("debug")("sm:slave:preroller")

module.exports = class Preroller
    constructor: (@stream,@key,@uri,@transcode_uri,cb) ->
        @_counter = 1

        # -- need to look at the stream to get characteristics -- #

        @stream.log.debug "Preroller calling getStreamKey"

        @stream.getStreamKey (@streamKey) =>
            @stream.log.debug "Preroller: Stream key is #{@streamKey}. Ready to start serving."

        cb? null, @

    #----------

    pump: (client,socket,writer,cb) ->
        cb = _.once(cb)
        aborted = false
        # short-circuit if we haven't gotten a stream key yet
        if !@streamKey || !@uri
            cb new Error("Preroll request before streamKey or missing URI.")
            return true

        # short-circuit if the socket has already disconnected
        if socket.destroyed
            cb new Error("Preroll request got destroyed socket.")
            return true

        count = @_counter++

        # If the preroll request can't be made in 5 seconds or less,
        # abort the preroll.
        # TODO: Make the timeout wait configurable
        prerollTimeout = setTimeout(=>
            @stream.log.debug "preroll request timeout. Aborting.", count
            req.abort()
            aborted = true
            detach new Error("Preroll request timed out.")
        , 5*1000)

        # -- Set up our ad URI -- #

        uri = @uri
            .replace("!KEY!", @streamKey)
            .replace("!IP!", client.ip)
            .replace("!STREAM!", @key)
            .replace("!UA!", encodeURIComponent(client.ua))
            .replace("!UUID!", client.session_id)

        debug "Ad request URI is #{uri}"

        # -- make a request to the ad server -- #

        treq = null
        adreq = request.get uri, (err,res,body) =>
            if err
                perr = new Error "Ad request returned error: #{err}"
                @stream.log.error perr, error:err
                debug perr
                return detach perr

            if res.statusCode == 200
                new Preroller.AdObject body, (err,obj) =>
                    if err
                        @stream.log.debug "Ad request was unsuccessful: #{err}"
                        return detach err

                    if obj.creativeURL
                        # we need to take the creative URL and pass it off
                        # to the transcoder
                        debug "Preparing transcoder request for #{obj.creativeURL} with key #{@streamKey}."

                        treq = request.get(@transcode_uri, qs:{ uri:obj.creativeURL, key:@streamKey })
                            .once "response", (resp) =>
                                debug "Transcoder response received: #{resp.statusCode}"

                                if resp.statusCode == 200
                                    treq.pipe(writer,end:false)
                                    treq.once "end", =>
                                        debug "Transcoder pipe complete."

                                        # we return by giving a function that should be called when the
                                        # impression criteria have been met
                                        detach null, =>
                                            if obj.impressionURL
                                                request.get obj.impressionURL, (err,resp,body) =>
                                                    if err
                                                        @stream.log.error "Failed to hit impression URL #{obj.impressionURL}: #{err}"
                                                    else
                                                        debug "Impression URL hit successfully."
                                            else
                                                @stream.log.debug "Session reached preroll impression criteria, but no impression URL present."
                                                debug "No impression URL found."
                                else
                                    err = new Error "Non-200 response from transcoder."
                                    debug err
                                    detach err
                            .once "error", (err) =>
                                debug "Transcoder request error: #{err}"
                                detach err

                    else
                        # no creative means just send the client on their way
                        detach()

            else
                perr = new Error "Ad request returned non-200 response: #{body}"
                @stream.log.debug perr
                return detach perr

        detach = (err,impcb) =>
            clearTimeout(prerollTimeout) if prerollTimeout
            socket.removeListener "close", conn_pre_abort
            socket.removeListener "end", conn_pre_abort
            cb err, impcb

        # attach a close listener to the response, to be fired if it gets
        # shut down and we should abort the request

        conn_pre_abort = =>
            detach()
            if socket.destroyed
                @stream.log.debug "aborting preroll ", count
                adreq?.abort()
                treq?.abort()
                aborted = true

        socket.once "close", conn_pre_abort
        socket.once "end", conn_pre_abort

    #----------

    class @AdObject
        constructor: (xmldoc,cb) ->
            @creativeURL    = null
            @impressionURL  = null

            @doc            = null

            debug "Parsing ad object XML"

            doc = new xmldom.DOMParser().parseFromString(xmldoc)

            debug "XML doc parsed."

            # -- VAST Support -- #

            if xpath.select("/VAST",doc)
                debug "VAST wrapper detected"

                if ad = xpath.select("VAST/Ad/InLine",doc)?[0]<commit_msg>Preroller: Include request count in debugging messages<commit_after>_       = require "underscore"
http    = require "http"
url     = require "url"
request = require "request"
xmldom  = require "xmldom"
xpath   = require "xpath"

debug = require("debug")("sm:slave:preroller")

module.exports = class Preroller
    constructor: (@stream,@key,@uri,@transcode_uri,cb) ->
        @_counter = 1

        # -- need to look at the stream to get characteristics -- #

        @stream.log.debug "Preroller calling getStreamKey"

        @stream.getStreamKey (@streamKey) =>
            @stream.log.debug "Preroller: Stream key is #{@streamKey}. Ready to start serving."

        cb? null, @

    #----------

    pump: (client,socket,writer,cb) ->
        cb = _.once(cb)
        aborted = false
        # short-circuit if we haven't gotten a stream key yet
        if !@streamKey || !@uri
            cb new Error("Preroll request before streamKey or missing URI.")
            return true

        # short-circuit if the socket has already disconnected
        if socket.destroyed
            cb new Error("Preroll request got destroyed socket.")
            return true

        count = @_counter++

        pdebug = (msg,args...) ->
            debug "#{count}: #{msg}", args...

        # If the preroll request can't be made in 5 seconds or less,
        # abort the preroll.
        # TODO: Make the timeout wait configurable
        prerollTimeout = setTimeout(=>
            @stream.log.debug "preroll request timeout. Aborting.", count
            pdebug "Hit timeout. Triggering abort."
            req.abort()
            aborted = true
            detach new Error("Preroll request timed out.")
        , 5*1000)

        # -- Set up our ad URI -- #

        uri = @uri
            .replace("!KEY!", @streamKey)
            .replace("!IP!", client.ip)
            .replace("!STREAM!", @key)
            .replace("!UA!", encodeURIComponent(client.ua))
            .replace("!UUID!", client.session_id)

        pdebug "Ad request URI is #{uri}"

        # -- make a request to the ad server -- #

        treq = null
        adreq = request.get uri, (err,res,body) =>
            if err
                perr = new Error "Ad request returned error: #{err}"
                @stream.log.error perr, error:err
                pdebug perr
                return detach perr

            if res.statusCode == 200
                new Preroller.AdObject body, (err,obj) =>
                    if err
                        perr = "Ad request was unsuccessful: #{err}"
                        @stream.log.debug perr
                        pdebug perr
                        return detach err

                    if obj.creativeURL
                        # we need to take the creative URL and pass it off
                        # to the transcoder
                        pdebug "Preparing transcoder request for #{obj.creativeURL} with key #{@streamKey}."

                        treq = request.get(@transcode_uri, qs:{ uri:obj.creativeURL, key:@streamKey })
                            .once "response", (resp) =>
                                pdebug "Transcoder response received: #{resp.statusCode}"

                                if resp.statusCode == 200
                                    treq.pipe(writer,end:false)
                                    treq.once "end", =>
                                        pdebug "Transcoder pipe complete."

                                        # we return by giving a function that should be called when the
                                        # impression criteria have been met
                                        detach null, =>
                                            if obj.impressionURL
                                                request.get obj.impressionURL, (err,resp,body) =>
                                                    if err
                                                        @stream.log.error "Failed to hit impression URL #{obj.impressionURL}: #{err}"
                                                    else
                                                        pdebug "Impression URL hit successfully."
                                            else
                                                @stream.log.debug "Session reached preroll impression criteria, but no impression URL present."
                                                pdebug "No impression URL found."
                                else
                                    err = new Error "Non-200 response from transcoder."
                                    pdebug err
                                    detach err
                            .once "error", (err) =>
                                pdebug "Transcoder request error: #{err}"
                                detach err

                    else
                        # no creative means just send the client on their way
                        detach()

            else
                perr = new Error "Ad request returned non-200 response: #{body}"
                @stream.log.debug perr
                pdebug perr
                return detach perr

        detach = _.once (err,impcb) =>
            pdebug "In detach"
            clearTimeout(prerollTimeout) if prerollTimeout
            socket.removeListener "close", conn_pre_abort
            socket.removeListener "end", conn_pre_abort
            cb err, impcb

        # attach a close listener to the response, to be fired if it gets
        # shut down and we should abort the request

        conn_pre_abort = =>
            detach()
            if socket.destroyed
                pdebug "Aborting"
                @stream.log.debug "aborting preroll ", count
                adreq?.abort()
                treq?.abort()
                aborted = true

        socket.once "close", conn_pre_abort
        socket.once "end", conn_pre_abort

    #----------

    class @AdObject
        constructor: (xmldoc,cb) ->
            @creativeURL    = null
            @impressionURL  = null

            @doc            = null

            debug "Parsing ad object XML"

            doc = new xmldom.DOMParser().parseFromString(xmldoc)

            debug "XML doc parsed."

            # -- VAST Support -- #

            if xpath.select("/VAST",doc)
                debug "VAST wrapper detected"

                if ad = xpath.select("VAST/Ad/InLine",doc)?[0]
{'fraction_non_alphanumeric': 0.08453051829510279, 'fraction_numerical': 0.0031748551472339077, 'mean_word_length': 1.1612349914236706, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 32, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 10}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 60, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1355653', 'n_tokens_mistral': 3057, 'n_tokens_neox': 2823, 'n_words': 936}