Home Explore Help
Register Sign In
J1X
/
huginn
Watch 2
Star 0
Fork 0

Browse Source
add DelayAgent for buffering incoming Events

Andrew Cantino 10 years ago
parent
9c5847451b
commit
2d72e18fef
3 changed files with 179 additions and 2 deletions
Show Diff Stats
  1. 65 0
      app/models/agents/delay_agent.rb
  2. 2 2
      config/initializers/delayed_job.rb
  3. 112 0
      spec/models/agents/delay_agent_spec.rb

+ 65 - 0
app/models/agents/delay_agent.rb
View File 

@@ -0,0 +1,65 @@
1 
+module Agents
2 
+  class DelayAgent < Agent
3 
+    default_schedule "every_12h"
4 
+
5 
+    description <<-MD
6 
+      The DelayAgent stores received Events and emits copies of them on a schedule. Use this as a buffer or queue of Events.
7 
+
8 
+      `max_events` should be set to the maximum number of events that you'd like to hold in the buffer. When this number is
9 
+      reached, new events will either be ignored, or will displace the oldest event already in the buffer, depending on
10 
+      whether you set `keep` to `newest` or `oldest`.
11 
+
12 
+      `expected_receive_period_in_days` is used to determine if the Agent is working. Set it to the maximum number of days
13 
+      that you anticipate passing without this Agent receiving an incoming Event.
14 
+    MD
15 
+
16 
+    def default_options
17 
+      {
18 
+        'expected_receive_period_in_days' => "10",
19 
+        'max_events' => "100",
20 
+        'keep' => 'newest'
21 
+      }
22 
+    end
23 
+
24 
+    def validate_options
25 
+      unless options['expected_receive_period_in_days'].present? && options['expected_receive_period_in_days'].to_i > 0
26 
+        errors.add(:base, "Please provide 'expected_receive_period_in_days' to indicate how many days can pass before this Agent is considered to be not working")
27 
+      end
28 
+
29 
+      unless options['keep'].present? && options['keep'].in?(%w[newest oldest])
30 
+        errors.add(:base, "The 'keep' option is required and must be set to 'oldest' or 'newest'")
31 
+      end
32 
+
33 
+      unless options['max_events'].present? && options['max_events'].to_i > 0
34 
+        errors.add(:base, "The 'max_events' option is required and must be an integer greater than 0")
35 
+      end
36 
+    end
37 
+
38 
+    def working?
39 
+      last_receive_at && last_receive_at > options['expected_receive_period_in_days'].to_i.days.ago && !recent_error_logs?
40 
+    end
41 
+
42 
+    def receive(incoming_events)
43 
+      incoming_events.each do |event|
44 
+        memory['event_ids'] ||= []
45 
+        memory['event_ids'] << event.id
46 
+        if memory['event_ids'].length > interpolated['max_events'].to_i
47 
+          if interpolated['keep'] == 'newest'
48 
+            memory['event_ids'].shift
49 
+          else
50 
+            memory['event_ids'].pop
51 
+          end
52 
+        end
53 
+      end
54 
+    end
55 
+
56 
+    def check
57 
+      if memory['event_ids'] && memory['event_ids'].length > 0
58 
+        received_events.where(id: memory['event_ids']).reorder('events.id asc').each do |event|
59 
+          create_event payload: event.payload
60 
+        end
61 
+        memory['event_ids'] = []
62 
+      end
63 
+    end
64 
+  end
65 
+end

+ 2 - 2
config/initializers/delayed_job.rb
View File 

@@ -6,8 +6,8 @@ Delayed::Worker.default_priority = 10
6 6 
 Delayed::Worker.delay_jobs = !Rails.env.test?
7 7 
 Delayed::Worker.sleep_delay = (ENV['DELAYED_JOB_SLEEP_DELAY'].presence || 10).to_f
8 8
9 
-# Delayed::Worker.logger = Logger.new(Rails.root.join('log', 'delayed_job.log'))
10 
-# Delayed::Worker.logger.level = Logger::DEBUG
9 
+Delayed::Worker.logger = Logger.new(STDOUT)
10 
+Delayed::Worker.logger.level = Logger::WARN
11 11
12 12 
 class Delayed::Job
13 13 
   scope :pending, ->{ where("locked_at IS NULL AND attempts = 0") }

+ 112 - 0
spec/models/agents/delay_agent_spec.rb
View File 

@@ -0,0 +1,112 @@
1 
+require 'spec_helper'
2 
+
3 
+describe Agents::DelayAgent do
4 
+  let(:agent) do
5 
+    _agent = Agents::DelayAgent.new(name: 'My DelayAgent')
6 
+    _agent.options = _agent.default_options.merge('max_events' => 2)
7 
+    _agent.user = users(:bob)
8 
+    _agent.sources << agents(:bob_website_agent)
9 
+    _agent.save!
10 
+    _agent
11 
+  end
12 
+
13 
+  def create_event
14 
+    _event = Event.new(payload: { random: rand })
15 
+    _event.agent = agents(:bob_website_agent)
16 
+    _event.save!
17 
+    _event
18 
+  end
19 
+
20 
+  let(:first_event) { create_event }
21 
+  let(:second_event) { create_event }
22 
+  let(:third_event) { create_event }
23 
+
24 
+  describe "#working?" do
25 
+    it "checks if events have been received within expected receive period" do
26 
+      expect(agent).not_to be_working
27 
+      Agents::DelayAgent.async_receive agent.id, [events(:bob_website_agent_event).id]
28 
+      expect(agent.reload).to be_working
29 
+      the_future = (agent.options[:expected_receive_period_in_days].to_i + 1).days.from_now
30 
+      stub(Time).now { the_future }
31 
+      expect(agent.reload).not_to be_working
32 
+    end
33 
+  end
34 
+
35 
+  describe "validation" do
36 
+    before do
37 
+      expect(agent).to be_valid
38 
+    end
39 
+
40 
+    it "should validate max_events" do
41 
+      agent.options.delete('max_events')
42 
+      expect(agent).not_to be_valid
43 
+      agent.options['max_events'] = ""
44 
+      expect(agent).not_to be_valid
45 
+      agent.options['max_events'] = "0"
46 
+      expect(agent).not_to be_valid
47 
+      agent.options['max_events'] = "10"
48 
+      expect(agent).to be_valid
49 
+    end
50 
+
51 
+    it "should validate presence of expected_receive_period_in_days" do
52 
+      agent.options['expected_receive_period_in_days'] = ""
53 
+      expect(agent).not_to be_valid
54 
+      agent.options['expected_receive_period_in_days'] = 0
55 
+      expect(agent).not_to be_valid
56 
+      agent.options['expected_receive_period_in_days'] = -1
57 
+      expect(agent).not_to be_valid
58 
+    end
59 
+
60 
+    it "should validate keep" do
61 
+      agent.options.delete('keep')
62 
+      expect(agent).not_to be_valid
63 
+      agent.options['keep'] = ""
64 
+      expect(agent).not_to be_valid
65 
+      agent.options['keep'] = 'wrong'
66 
+      expect(agent).not_to be_valid
67 
+      agent.options['keep'] = 'newest'
68 
+      expect(agent).to be_valid
69 
+      agent.options['keep'] = 'oldest'
70 
+      expect(agent).to be_valid
71 
+    end
72 
+  end
73 
+
74 
+  describe "#receive" do
75 
+    it "records Events" do
76 
+      expect(agent.memory).to be_empty
77 
+      agent.receive([first_event])
78 
+      expect(agent.memory).not_to be_empty
79 
+      agent.receive([second_event])
80 
+      expect(agent.memory['event_ids']).to eq [first_event.id, second_event.id]
81 
+    end
82 
+
83 
+    it "keeps the newest when 'keep' is set to 'newest'" do
84 
+      expect(agent.options['keep']).to eq 'newest'
85 
+      agent.receive([first_event, second_event, third_event])
86 
+      expect(agent.memory['event_ids']).to eq [second_event.id, third_event.id]
87 
+    end
88 
+
89 
+    it "keeps the oldest when 'keep' is set to 'oldest'" do
90 
+      agent.options['keep'] = 'oldest'
91 
+      agent.receive([first_event, second_event, third_event])
92 
+      expect(agent.memory['event_ids']).to eq [first_event.id, second_event.id]
93 
+    end
94 
+  end
95 
+
96 
+  describe "#check" do
97 
+    it "re-emits Events and clears the memory" do
98 
+      agent.receive([first_event, second_event, third_event])
99 
+      expect(agent.memory['event_ids']).to eq [second_event.id, third_event.id]
100 
+
101 
+      expect {
102 
+        agent.check
103 
+      }.to change { agent.events.count }.by(2)
104 
+
105 
+      events = agent.events.reorder('events.id desc')
106 
+      expect(events.first.payload).to eq third_event.payload
107 
+      expect(events.second.payload).to eq second_event.payload
108 
+
109 
+      expect(agent.memory['event_ids']).to eq []
110 
+    end
111 
+  end
112 
+end