# encoding: utf-8

require 'thread'
require 'one_apm/support/event/timer'

module OneApm
  module Support
    class EventLoop

      def initialize
        @self_pipe_rd, @self_pipe_wr = IO.pipe
        @event_queue = Queue.new
        @stopped     = false
        @timers      = {}

        @subscriptions = Hash.new { |h,k| h[k] = [] }
        @subscriptions[:__add_timer] << Proc.new { |t| set_timer(t) }
        @subscriptions[:__add_event] << Proc.new { |e, blk| @subscriptions[e] << blk }
      end

      def set_timer(timer)
        existing_timer = @timers[timer.event]

        if existing_timer
          elapsed_interval = Time.now - existing_timer.last_interval_start
          timer.advance(elapsed_interval)
        end

        @timers[timer.event] = timer

        fire_timer(timer)
      end

      def next_timeout
        return nil if @timers.empty?
        timeout = @timers.values.map(&:next_fire_time).min - Time.now
        timeout < 0 ? 0 : timeout
      end

      def stopped?
        @stopped
      end

      def stop
        @stopped = true
        wakeup
      end

      def run
        OneApm::Manager.logger.debug "Running event loop"
        while !stopped?
          run_once
        end
      end

      def run_once(nonblock=false)
        wait_to_run(nonblock)

        prune_timers
        fire_timers

        until @event_queue.empty?
          evt, args = @event_queue.pop
          dispatch_event(evt, args)
          reschedule_timer_for_event(evt)
        end
      end

      def wait_to_run(nonblock)
        timeout = nonblock ? 0 : next_timeout
        ready = IO.select([@self_pipe_rd], nil, nil, timeout)

        if ready && ready[0] && ready[0][0] && ready[0][0] == @self_pipe_rd
          @self_pipe_rd.read(1)
        end
      end

      def fire_timers
        @timers.each do |event, timer|
          fire_timer(timer)
        end
      end

      def fire_timer(timer)
        if timer.due?
          @event_queue << [timer.event]
          timer.set_fired_time
        end
      end

      def prune_timers
        @timers.delete_if { |e, t| t.finished? }
      end

      def dispatch_event(event, args)
        OneApm::Manager.logger.debug("EventLoop: Dispatching event '#{event}' with #{@subscriptions[event].size} callback(s).")

        errors = []
        @subscriptions[event].each do |s|
          begin
            s.call(*args)
          rescue OneApm::ForceRestartException, OneApm::ForceDisconnectException
            raise
          rescue => e
            errors << e
          end
        end

        if !errors.empty?
          OneApm::Manager.logger.error "#{errors.size} error(s) running task for event '#{event}' in Agent Event Loop:", *errors
        end
      end

      def reschedule_timer_for_event(e)
        @timers[e].reschedule if @timers[e]
      end

      def on(event, &blk)
        fire(:__add_event, event, blk)
      end

      def fire(event, *args)
        @event_queue << [event, args]
        wakeup
      end

      def fire_every(interval, event)
        OneApm::Manager.logger.debug "Firing event #{event} every #{interval} seconds."
        fire(:__add_timer, Timer.new(interval, event, true))
      end

      def fire_after(interval, event)
        OneApm::Manager.logger.debug "Firing event #{event} after #{interval} seconds."
        fire(:__add_timer, Timer.new(interval, event, false))
      end

      def wakeup
        @self_pipe_wr.write_nonblock '.'
      rescue Errno::EAGAIN
        OneApm::Manager.logger.debug "Failed to wakeup event loop"
      end
    end
  end
end