RESTful polling

Gemfile

@@ -36,6 +36,7 @@ gem "twitter"
 gem 'twitter-stream', '>=0.1.16'
 gem 'em-http-request'
 gem 'weibo_2'
+gem 'cobravsmongoose'
 
 platforms :ruby_18 do
   gem 'system_timer'

Gemfile.lock

@@ -40,6 +40,7 @@ GEM
       kaminari (>= 0.13)
       rails (>= 3.1)
     builder (3.0.4)
+    cobravsmongoose (0.0.2)
     coderay (1.0.9)
     coffee-rails (3.2.2)
       coffee-script (>= 2.2.0)
@@ -270,6 +271,7 @@ DEPENDENCIES
   better_errors
   binding_of_caller
   bootstrap-kaminari-views
+  cobravsmongoose
   coffee-rails (~> 3.2.1)
   coveralls
   daemons

app/models/agent.rb

@@ -77,6 +77,10 @@ def receive_webhook(params)
     ["not implemented", 404]
   end
 
+  def em_start
+    # implement me in your subclass of Agent if continuous.
+  end
+
   # Implement me in your subclass to decide if your Agent is working.
   def working?
     raise "Implement me in your subclass"
@@ -153,6 +157,10 @@ def can_create_events?
     !cannot_create_events?
   end
 
+  def continuous?
+    self.class.continuous?
+  end
+
   def log(message, options = {})
     puts "Agent##{id}: #{message}" unless Rails.env.test?
     AgentLog.log_for_agent(self, message, options)
@@ -229,6 +237,14 @@ def cannot_receive_events?
       !!@cannot_receive_events
     end
 
+    def continuous!
+      @continuous = true
+    end
+
+    def continuous?
+      !!@continuous
+    end
+
     # 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.

app/models/agents/rest_polling_agent.rb

@@ -0,0 +1,65 @@
+require 'em-http'
+require 'cobravsmongoose'
+require 'json'
+
+module Agents
+  class RestPollingAgent < Agent
+    cannot_be_scheduled!
+    cannot_receive_events!
+    continuous!
+
+    description <<-MD
+      Does long polling on a rest API url. The resulting type should be json or XML (which will be translated to json).
+    MD
+
+    event_description "REST event"
+
+    def default_options
+      {  
+          'url' => "http://some.polling/url",
+          'type' => "xml"
+      }
+    end
+
+    def em_start
+      t = Time.now
+      http = EM::HttpRequest.new(options['url']).get
+      http.errback {
+          if Time.now - t > 0.1
+            em_start
+          else
+            EM.add_timer(1) {
+                em_start
+              }
+          end
+        }
+      http.callback {
+          puts http.response
+          if options['type'] == "json"
+            pld = JSON.parse(http.response)
+          else
+            pld = CobraVsMongoose.xml_to_hash(http.response)
+          end
+          create_event :payload => pld
+          if Time.now - t > 0.1
+            em_start
+          else
+            EM.add_timer(1) {
+                em_start
+              }
+          end
+        }
+    end
+
+    def working?
+      true
+    end
+
+    def validate_options
+      errors.add(:base, "url and type are required") unless options['type'].present? && options['url'].present?
+      if options['type'] != 'json' && options['type'] != 'xml'
+        errors.add(:base, "unrecognized type")
+      end
+    end
+  end
+end

bin/schedule.rb

@@ -12,6 +12,8 @@
 end
 
 require 'rufus/scheduler'
+require 'eventmachine'
+require 'set'
 
 class HuginnScheduler
   attr_accessor :mutex
@@ -87,9 +89,23 @@ def run!
       end
     end
 
+    active_agents = Set.new
+    EM.run do
+        EM.add_periodic_timer(10) {
+            Agent.find_each do |agent|
+                   if agent.continuous? 
+                     if active_agents.add? agent
+                       puts "Starting continuous agent"
+                       agent.em_start
+                     end
+                   end
+                 end
+          }
+      end
+
     rufus_scheduler.join
   end
 end
 
 scheduler = HuginnScheduler.new
-scheduler.run!
+scheduler.run!

db/schema.rb

@@ -11,7 +11,7 @@
 #
 # It's strongly recommended to check this file into your version control system.
 
-ActiveRecord::Schema.define(:version => 20131227000021) do
+ActiveRecord::Schema.define(:version => 20140127164931) do
 
   create_table "agent_logs", :force => true do |t|
     t.integer  "agent_id",                         :null => false
@@ -47,17 +47,17 @@
   add_index "agents", ["user_id", "created_at"], :name => "index_agents_on_user_id_and_created_at"
 
   create_table "delayed_jobs", :force => true do |t|
-    t.integer  "priority",   :default => 0
-    t.integer  "attempts",   :default => 0
-    t.text     "handler"
+    t.integer  "priority",                       :default => 0
+    t.integer  "attempts",                       :default => 0
+    t.text     "handler",    :limit => 16777215
     t.text     "last_error"
     t.datetime "run_at"
     t.datetime "locked_at"
     t.datetime "failed_at"
     t.string   "locked_by"
     t.string   "queue"
-    t.datetime "created_at",                :null => false
-    t.datetime "updated_at",                :null => false
+    t.datetime "created_at",                                    :null => false
+    t.datetime "updated_at",                                    :null => false
   end
 
   add_index "delayed_jobs", ["priority", "run_at"], :name => "delayed_jobs_priority"