app/models/agent.rb
# Find all Agents that have received Events since the last execution of this method.  Update those Agents with
# their new `last_checked_event_id` and queue each of the Agents to be called with #receive using `async_receive`.
# This is called by bin/schedule.rb periodically.
Find all Agents that have received Events since the last execution of this method.  Update those Agents with
their new `last_checked_event_id` and queue each of the Agents to be called with #receive using `async_receive`.
This is called by bin/schedule.rb periodically.
Find all Agents that have received Events since the last execution of this method . Update those Agents with their new ` last_checked_event_id ` and queue each of the Agents to be called with #receive using ` async_receive ` . This is called by bin / schedule . rb periodically .
def receive!(options={})
      Agent.transaction do
        scope = Agent.
                select("agents.id AS receiver_agent_id, sources.id AS source_agent_id, events.id AS event_id").
                joins("JOIN links ON (links.receiver_id = agents.id)").
                joins("JOIN agents AS sources ON (links.source_id = sources.id)").
                joins("JOIN events ON (events.agent_id = sources.id AND events.id > links.event_id_at_creation)").
                where("agents.last_checked_event_id IS NULL OR events.id > agents.last_checked_event_id")
        if options[:only_receivers].present?
          scope = scope.where("agents.id in (?)", options[:only_receivers])
        end
        sql = scope.to_sql()
        agents_to_events = {}
        Agent.connection.select_rows(sql).each do |receiver_agent_id, source_agent_id, event_id|
          agents_to_events[receiver_agent_id.to_i] ||= []
          agents_to_events[receiver_agent_id.to_i] << event_id
        end
        event_ids = agents_to_events.values.flatten.uniq.compact
        Agent.where(:id => agents_to_events.keys).each do |agent|
          agent.update_attribute :last_checked_event_id, event_ids.max
          Agent.async_receive(agent.id, agents_to_events[agent.id].uniq)
        end
        {
          :agent_count => agents_to_events.keys.length,
          :event_count => event_ids.length
        }
      end
    end
def receive! ( options = { } ) Agent . transaction do scope = Agent . select ( "agents.id AS receiver_agent_id, sources.id AS source_agent_id, events.id AS event_id" ) . joins ( "JOIN links ON (links.receiver_id = agents.id)" ) . joins ( "JOIN agents AS sources ON (links.source_id = sources.id)" ) . joins ( "JOIN events ON (events.agent_id = sources.id AND events.id > links.event_id_at_creation)" ) . where ( "agents.last_checked_event_id IS NULL OR events.id > agents.last_checked_event_id" ) if options [ :only_receivers ] . present? scope = scope . where ( "agents.id in (?)" , options [ :only_receivers ] ) end sql = scope . to_sql ( ) agents_to_events = { } Agent . connection . select_rows ( sql ) . each do | receiver_agent_id , source_agent_id , event_id | agents_to_events [ receiver_agent_id . to_i ] ||= [ ] agents_to_events [ receiver_agent_id . to_i ] << event_id end event_ids = agents_to_events . values . flatten . uniq . compact Agent . where ( :id => agents_to_events . keys ) . each do | agent | agent . update_attribute :last_checked_event_id , event_ids . max Agent . async_receive ( agent . id , agents_to_events [ agent . id ] . uniq ) end { :agent_count => agents_to_events . keys . length , :event_count => event_ids . length } end end
Find all Agents that have received Events since the last execution of this method.
Find all Agents that have received Events since the last execution of this method .
options