lib/powertrack/streaming/stream.rb
def track_once(options, retrier)
      logger.info "Starting tracker for retry ##{retrier.retries}..."
      backfill_minutes = options[:backfill_minutes]
      stop_timeout = options[:stop_timeout]
      on_heartbeat = options[:on_heartbeat]
      on_message = options[:on_message]
      on_activity = options[:on_activity]
      on_system = options[:on_system]
      close_now = options[:close_now] || lambda { false }

      buffer = PowerTrack::DataBuffer.new
      closed = false
      disconnected = false
      resp_status = DEFAULT_OK_RESPONSE_STATUS
      resp_error = nil
      resp_body = nil

      EM.run do
        logger.info "Starting the reactor..."
        con = connect('stream')
        get_opts = {
          head: track_req_headers(options[:compressed]),
          query: {}
        }

        # add a timeframe in replay mode
        if @replay
          now = Time.now
          # start 1 hour ago by default
          from = options[:from] || (now - 60*60)
          # stop 30 minutes ago by default
          to = options[:to] || (now - 30*60)

          get_opts[:query].merge!({
            'fromDate' => from.utc.strftime(REPLAY_TIMESTAMP_FORMAT),
            'toDate' => to.utc.strftime(REPLAY_TIMESTAMP_FORMAT)
          })

          logger.info "Replay mode enabled from '#{from}' to '#{to}'"
        end

        if backfill_minutes
          get_opts[:query]['backfillMinutes'] = backfill_minutes
        end

        http = con.get(get_opts)

        # polls to see if the connection should be closed
        close_watcher = EM.add_periodic_timer(1) do
          # exit if required
          if close_now.call
            logger.info "Time to close the tracker"
            closed = true
            close_watcher.cancel
            con.close
          end
        end

        # simulate periodic disconnections
        if options[:fake_disconnections]
           EM.add_timer(rand(options[:fake_disconnections])) do
             con.close
           end
        end

        http.stream do |chunk|
          # ignore data if already disconnected, thus avoiding synchronizing the
          # buffer. Nevertheless, this should never happen...
          # TODO: log a warning if it happens

          if disconnected
            logger.warn "Message received while already disconnected"
            next
          end

          # process the chunk
          buffer.process(chunk) do |raw|
            logger.debug "New message received"

            # get the message type and its (optional) level
            m_type, m_level = message_type(raw)

            # reset retries when some (valid) data are received but not in replay
            # mode where we don't want to retry on the same timeframe again and
            # again when GNIP periodically fails
            if !@replay && retrier.retrying? && m_level != :error
              logger.info "Resetting retries..."
              retrier.reset!
            end

            EM.defer do
              # select the right message handler(s) according to the message type
              if m_type == :heartbeat
                on_heartbeat.call if on_heartbeat
              else
                # JSON decoding if required
                message = options[:raw] ? raw : MultiJson.decode(raw)

                on_message.call(message) if on_message

                case m_type
                when :system then on_system.call(message) if on_system
                when :activity then on_activity.call(message) if on_activity
                end
              end

              # TODO: manage exceptions at this level
            end
          end
        end

        # reconnection on error
        reconnect_cb = lambda do |http_client|
          logger.info "Disconnected after #{retrier.retries} retries"
          disconnected = true

          resp_status = http_client.response_header.status

          # stop the stream if required so or the replay is simply over
          if closed || (@replay && resp_status == DEFAULT_OK_RESPONSE_STATUS)
            # close immediately if required
            wait_til_defers_finish_and_stop(stop_timeout)
            # tell the retrier the tracking is over
            retrier.stop
          else
            # cancel the periodic close watcher
            close_watcher.cancel

            resp_status ||= DEFAULT_OK_RESPONSE_STATUS
            resp_error = http_client.error
            resp_body = http_client.response

            wait_til_defers_finish_and_stop(stop_timeout)
          end
        end

        http.callback(&reconnect_cb)
        http.errback(&reconnect_cb)
      end

      [ resp_status, resp_error, resp_body ]
    end
# Connects to the /track endpoint.
Connects to the /track endpoint.
Connects to the / track endpoint .
def track_once(options, retrier)
      logger.info "Starting tracker for retry ##{retrier.retries}..."
      backfill_minutes = options[:backfill_minutes]
      stop_timeout = options[:stop_timeout]
      on_heartbeat = options[:on_heartbeat]
      on_message = options[:on_message]
      on_activity = options[:on_activity]
      on_system = options[:on_system]
      close_now = options[:close_now] || lambda { false }
      buffer = PowerTrack::DataBuffer.new
      closed = false
      disconnected = false
      resp_status = DEFAULT_OK_RESPONSE_STATUS
      resp_error = nil
      resp_body = nil
      EM.run do
        logger.info "Starting the reactor..."
        con = connect('stream')
        get_opts = {
          head: track_req_headers(options[:compressed]),
          query: {}
        }
        if @replay
          now = Time.now
          from = options[:from] || (now - 60*60)
          to = options[:to] || (now - 30*60)
          get_opts[:query].merge!({
            'fromDate' => from.utc.strftime(REPLAY_TIMESTAMP_FORMAT),
            'toDate' => to.utc.strftime(REPLAY_TIMESTAMP_FORMAT)
          })
          logger.info "Replay mode enabled from '#{from}' to '#{to}'"
        end
        if backfill_minutes
          get_opts[:query]['backfillMinutes'] = backfill_minutes
        end
        http = con.get(get_opts)
        close_watcher = EM.add_periodic_timer(1) do
          if close_now.call
            logger.info "Time to close the tracker"
            closed = true
            close_watcher.cancel
            con.close
          end
        end
        if options[:fake_disconnections]
           EM.add_timer(rand(options[:fake_disconnections])) do
             con.close
           end
        end
        http.stream do |chunk|
          if disconnected
            logger.warn "Message received while already disconnected"
            next
          end
          buffer.process(chunk) do |raw|
            logger.debug "New message received"
            m_type, m_level = message_type(raw)
            if !@replay && retrier.retrying? && m_level != :error
              logger.info "Resetting retries..."
              retrier.reset!
            end
            EM.defer do
              if m_type == :heartbeat
                on_heartbeat.call if on_heartbeat
              else
                message = options[:raw] ? raw : MultiJson.decode(raw)
                on_message.call(message) if on_message
                case m_type
                when :system then on_system.call(message) if on_system
                when :activity then on_activity.call(message) if on_activity
                end
              end
            end
          end
        end
        reconnect_cb = lambda do |http_client|
          logger.info "Disconnected after #{retrier.retries} retries"
          disconnected = true
          resp_status = http_client.response_header.status
          if closed || (@replay && resp_status == DEFAULT_OK_RESPONSE_STATUS)
            wait_til_defers_finish_and_stop(stop_timeout)
            retrier.stop
          else
            close_watcher.cancel
            resp_status ||= DEFAULT_OK_RESPONSE_STATUS
            resp_error = http_client.error
            resp_body = http_client.response
            wait_til_defers_finish_and_stop(stop_timeout)
          end
        end
        http.callback(&reconnect_cb)
        http.errback(&reconnect_cb)
      end
      [ resp_status, resp_error, resp_body ]
    end
def track_once ( options , retrier ) logger . info "Starting tracker for retry ##{retrier.retries}..." backfill_minutes = options [ :backfill_minutes ] stop_timeout = options [ :stop_timeout ] on_heartbeat = options [ :on_heartbeat ] on_message = options [ :on_message ] on_activity = options [ :on_activity ] on_system = options [ :on_system ] close_now = options [ :close_now ] || lambda { false } buffer = PowerTrack :: DataBuffer . new closed = false disconnected = false resp_status = DEFAULT_OK_RESPONSE_STATUS resp_error = nil resp_body = nil EM . run do logger . info "Starting the reactor..." con = connect ( 'stream' ) get_opts = { head : track_req_headers ( options [ :compressed ] ) , query : { } } if @replay now = Time . now from = options [ :from ] || ( now - 60 * 60 ) to = options [ :to ] || ( now - 30 * 60 ) get_opts [ :query ] . merge! ( { 'fromDate' => from . utc . strftime ( REPLAY_TIMESTAMP_FORMAT ) , 'toDate' => to . utc . strftime ( REPLAY_TIMESTAMP_FORMAT ) } ) logger . info "Replay mode enabled from '#{from}' to '#{to}'" end if backfill_minutes get_opts [ :query ] [ 'backfillMinutes' ] = backfill_minutes end http = con . get ( get_opts ) close_watcher = EM . add_periodic_timer ( 1 ) do if close_now . call logger . info "Time to close the tracker" closed = true close_watcher . cancel con . close end end if options [ :fake_disconnections ] EM . add_timer ( rand ( options [ :fake_disconnections ] ) ) do con . close end end http . stream do | chunk | if disconnected logger . warn "Message received while already disconnected" next end buffer . process ( chunk ) do | raw | logger . debug "New message received" m_type , m_level = message_type ( raw ) if ! @replay && retrier . retrying? && m_level != :error logger . info "Resetting retries..." retrier . reset! end EM . defer do if m_type == :heartbeat on_heartbeat . call if on_heartbeat else message = options [ :raw ] ? raw : MultiJson . decode ( raw ) on_message . call ( message ) if on_message case m_type when :system then on_system . call ( message ) if on_system when :activity then on_activity . call ( message ) if on_activity end end end end end reconnect_cb = lambda do | http_client | logger . info "Disconnected after #{retrier.retries} retries" disconnected = true resp_status = http_client . response_header . status if closed || ( @replay && resp_status == DEFAULT_OK_RESPONSE_STATUS ) wait_til_defers_finish_and_stop ( stop_timeout ) retrier . stop else close_watcher . cancel resp_status ||= DEFAULT_OK_RESPONSE_STATUS resp_error = http_client . error resp_body = http_client . response wait_til_defers_finish_and_stop ( stop_timeout ) end end http . callback ( & reconnect_cb ) http . errback ( & reconnect_cb ) end [ resp_status , resp_error , resp_body ] end
Connects to the /track endpoint.
Connects to the / track endpoint .
# add a timeframe in replay mode # start 1 hour ago by default # stop 30 minutes ago by default # polls to see if the connection should be closed # exit if required # simulate periodic disconnections # ignore data if already disconnected, thus avoiding synchronizing the # buffer. Nevertheless, this should never happen... # TODO: log a warning if it happens # process the chunk # get the message type and its (optional) level # reset retries when some (valid) data are received but not in replay # mode where we don't want to retry on the same timeframe again and # again when GNIP periodically fails # select the right message handler(s) according to the message type # JSON decoding if required # TODO: manage exceptions at this level # reconnection on error # stop the stream if required so or the replay is simply over # close immediately if required # tell the retrier the tracking is over # cancel the periodic close watcher
options
retrier