class Agent

# File lib/wattics-api-client/agent.rb, line 24
def self.dispose
  @@mutex.synchronize do
    unless @@instance.nil?
      @@instance.agent_thread_group.list.each(&:kill)
      @@instance = nil
    end
  end
end

# File lib/wattics-api-client/agent.rb, line 18
def self.get_instance(maximum_parallel_senders = 0)
  @@mutex.synchronize do
    @@instance ||= new(maximum_parallel_senders)
  end
end

# File lib/wattics-api-client/agent.rb, line 6
def initialize(maximum_parallel_senders = 0)
  @agent_thread_group = ThreadGroup.new
  @processor_pool = ProcessorPool.new(self, @agent_thread_group, maximum_parallel_senders)
  @enqueued_measurements_with_config = Hash.new { |h, k| h[k] = [] }
  @sent_measurements_with_context = BlockingQueue.new
  @measurement_sent_handler_list = Concurrent::Array.new
  start_processor_feeder
  start_measurement_sent_handler_dispatcher
  @wait_semaphore = Concurrent::Semaphore.new(0)
end

# File lib/wattics-api-client/agent.rb, line 111
def add_measurement_sent_handler
  @measurement_sent_handler_list << yield
end

# File lib/wattics-api-client/agent.rb, line 105
def report_sent_measurement(measurement, response)
  #Dont need to send to the measurement handler, <=400 erros are reported on the processor
  @sent_measurements_with_context << [measurement, response] if response.code <= 400
  @wait_semaphore.acquire
end

# File lib/wattics-api-client/agent.rb, line 80
def send(measurement, config)
  if measurement.is_a?(Array)
    @wait_semaphore.release(measurement.size)
    measurement_groups = measurement.group_by(&:id)
    measurement_groups.each do |channel_id, measurements_for_channel_id|
      measurements_with_config = measurements_for_channel_id.map { |measurement| MeasurementWithConfig.new(measurement, config) }
      @processor_already_bound_to_channel_id = @processor_pool.get_processor(channel_id)
      if @processor_already_bound_to_channel_id.nil?
        @enqueued_measurements_with_config[channel_id] += measurements_with_config
      else
        @processor_already_bound_to_channel_id.process(measurements_with_config)
      end
    end
  else
    @wait_semaphore.release
    measurement_with_config = MeasurementWithConfig.new(measurement, config)
    @processor_already_bound_to_channel_id = @processor_pool.get_processor(measurement.id)
    if @processor_already_bound_to_channel_id.nil?
      @enqueued_measurements_with_config[measurement.id] << measurement_with_config
    else
      @processor_already_bound_to_channel_id.process(measurement_with_config)
    end
  end
end

# File lib/wattics-api-client/agent.rb, line 76
def sleep_fix
  sleep 0.1
end

# File lib/wattics-api-client/agent.rb, line 61
def start_measurement_sent_handler_dispatcher
  @agent_thread_group.add(Thread.new do
    begin
      loop do
        array = @sent_measurements_with_context.pop
        next if array.nil?
        measurement = array[0]
        response = array[1]
        @measurement_sent_handler_list.each { |handler| handler.call(measurement, response) }
      end
    rescue ThreadError
    end
  end)
end

# File lib/wattics-api-client/agent.rb, line 39
def start_processor_feeder
  @agent_thread_group.add(Thread.new do
    begin
      loop do
        key, values = @enqueued_measurements_with_config.first
        if @enqueued_measurements_with_config.empty?
          sleep_fix
          next
        end
        processor = @processor_pool.get_processor(key)
        if processor.nil?
          sleep_fix
          next
        end
        @enqueued_measurements_with_config.delete(key)
        processor.process(values)
      end
    rescue ThreadError
    end
  end)
end

# File lib/wattics-api-client/agent.rb, line 33
def wait_until_last
  Thread.new do
    sleep 0.01 while @wait_semaphore.available_permits != 0
  end.join
end