DataOutputAgent: Add support for publishing via PubSubHubbub

A new option `push_hubs` is added, which lists PuSH hub endpoints to
publish updates to.

Author: knu

app/models/agents/data_output_agent.rb

@@ -1,5 +1,7 @@
 module Agents
   class DataOutputAgent < Agent
+    include WebRequestConcern
+
     cannot_be_scheduled!
 
     description  do
@@ -22,6 +24,7 @@ class DataOutputAgent < Agent
           * `template` - A JSON object representing a mapping between item output keys and incoming event values.  Use [Liquid](https://github.com/cantino/huginn/wiki/Formatting-Events-using-Liquid) to format the values.  Values of the `link`, `title`, `description` and `icon` keys will be put into the \\<channel\\> section of RSS output.  Value of the `self` key will be used as URL for this feed itself, which is useful when you serve it via reverse proxy.  The `item` key will be repeated for every Event.  The `pubDate` key for each item will have the creation time of the Event unless given.
           * `events_to_show` - The number of events to output in RSS or JSON. (default: `40`)
           * `ttl` - A value for the \\<ttl\\> element in RSS output. (default: `60`)
+          * `push_hubs` - Set to a list of PubSubHubbub endpoints you want to publish every update to. (default: none)  Note that publishing updates will make your feed public.  Popular hubs include [Superfeedr](https://pubsubhubbub.superfeedr.com/) and [Google](https://pubsubhubbub.appspot.com/).
 
         If you'd like to output RSS tags with attributes, such as `enclosure`, use something like the following in your `template`:
 
@@ -95,6 +98,29 @@ def validate_options
       unless options['template'].present? && options['template']['item'].present? && options['template']['item'].is_a?(Hash)
         errors.add(:base, "Please provide template and template.item")
       end
+
+      case options['push_hubs']
+      when nil
+      when Array
+        options['push_hubs'].each do |hub|
+          case hub
+          when /\{/
+            # Liquid templating
+          when String
+            begin
+              URI.parse(hub)
+            rescue URI::Error
+              errors.add(:base, "invalid URL found in push_hubs")
+              break
+            end
+          else
+            errors.add(:base, "push_hubs must be an array of endpoint URLs")
+            break
+          end
+        end
+      else
+        errors.add(:base, "push_hubs must be an array")
+      end
     end
 
     def events_to_show
@@ -130,6 +156,10 @@ def feed_description
       interpolated['template']['description'].presence || "A feed of Events received by the '#{name}' Huginn Agent"
     end
 
+    def push_hubs
+      interpolated['push_hubs'].presence || []
+    end
+
     def receive_web_request(params, method, format)
       unless interpolated['secrets'].include?(params['secret'])
         if format =~ /json/
@@ -160,40 +190,54 @@ def receive_web_request(params, method, format)
           interpolated
         end
 
+        now = Time.now
+
         if format =~ /json/
           content = {
             'title' => feed_title,
             'description' => feed_description,
-            'pubDate' => Time.now,
+            'pubDate' => now,
             'items' => simplify_item_for_json(items)
           }
 
           return [content, 200]
         else
-          content = Utils.unindent(<<-XML)
-            <?xml version="1.0" encoding="UTF-8" ?>
-            <rss version="2.0" xmlns:atom="http://www.w3.org/2005/Atom">
-            <channel>
-             <atom:link href=#{feed_url(secret: params['secret'], format: :xml).encode(xml: :attr)} rel="self" type="application/rss+xml" />
-             <atom:icon>#{feed_icon.encode(xml: :text)}</atom:icon>
-             <title>#{feed_title.encode(xml: :text)}</title>
-             <description>#{feed_description.encode(xml: :text)}</description>
-             <link>#{feed_link.encode(xml: :text)}</link>
-             <lastBuildDate>#{Time.now.rfc2822.to_s.encode(xml: :text)}</lastBuildDate>
-             <pubDate>#{Time.now.rfc2822.to_s.encode(xml: :text)}</pubDate>
-             <ttl>#{feed_ttl}</ttl>
-
+          hub_links = push_hubs.map { |hub|
+            <<-XML
+ <atom:link rel="hub" href=#{hub.encode(xml: :attr)}/>
+            XML
+          }.join
+
+          items = simplify_item_for_xml(items)
+                  .to_xml(skip_types: true, root: "items", skip_instruct: true, indent: 1)
+                  .gsub(%r{^</?items>\n}, '')
+
+          return [<<-XML, 200, 'text/xml']
+<?xml version="1.0" encoding="UTF-8" ?>
+<rss version="2.0" xmlns:atom="http://www.w3.org/2005/Atom">
+<channel>
+ <atom:link href=#{feed_url(secret: params['secret'], format: :xml).encode(xml: :attr)} rel="self" type="application/rss+xml" />
+ <atom:icon>#{feed_icon.encode(xml: :text)}</atom:icon>
+#{hub_links}
+ <title>#{feed_title.encode(xml: :text)}</title>
+ <description>#{feed_description.encode(xml: :text)}</description>
+ <link>#{feed_link.encode(xml: :text)}</link>
+ <lastBuildDate>#{now.rfc2822.to_s.encode(xml: :text)}</lastBuildDate>
+ <pubDate>#{now.rfc2822.to_s.encode(xml: :text)}</pubDate>
+ <ttl>#{feed_ttl}</ttl>
+#{items}
+</channel>
+</rss>
           XML
+        end
+      end
+    end
 
-          content += simplify_item_for_xml(items).to_xml(skip_types: true, root: "items", skip_instruct: true, indent: 1).gsub(/^<\/?items>/, '').strip
-
-          content += Utils.unindent(<<-XML)
-            </channel>
-            </rss>
-          XML
+    def receive(incoming_events)
+      url = feed_url(secret: interpolated['secrets'].first, format: :xml)
 
-          return [content, 200, 'text/xml']
-        end
+      push_hubs.each do |hub|
+        push_to_hub(hub, url)
       end
     end
 
@@ -262,5 +306,32 @@ def simplify_item_for_json(item)
         item
       end
     end
+
+    def push_to_hub(hub, url)
+      hub_uri =
+        begin
+          URI.parse(hub)
+        rescue URI::Error
+          nil
+        end
+
+      if !hub_uri.is_a?(URI::HTTP)
+        error "Invalid push endpoint: #{hub}"
+        return
+      end
+
+      log "Pushing #{url} to #{hub_uri}"
+
+      return if dry_run?
+
+      begin
+        faraday.post hub_uri, {
+          'hub.mode' => 'publish',
+          'hub.url' => url
+        }
+     rescue => e
+       error "Push failed: #{e.message}"
+      end
+    end
   end
 end

spec/models/agents/data_output_agent_spec.rb

@@ -73,6 +73,29 @@
     end
   end
 
+  describe "#receive" do
+    it "should push to hubs when push_hubs is given" do
+      agent.options[:push_hubs] = %w[http://push.example.com]
+      agent.options[:template] = { 'link' => 'http://huginn.example.org' }
+
+      alist = nil
+
+      stub_request(:post, 'http://push.example.com/')
+        .with(headers: { 'Content-Type' => %r{\Aapplication/x-www-form-urlencoded\s*(?:;|\z)} })
+        .to_return { |request|
+        alist = URI.decode_www_form(request.body).sort
+        { status: 200, body: 'ok' }
+      }
+
+      agent.receive(events(:bob_website_agent_event))
+
+      expect(alist).to eq [
+        ["hub.mode", "publish"],
+        ["hub.url", agent.feed_url(secret: agent.options[:secrets].first, format: :xml)]
+      ]
+    end
+  end
+
   describe "#receive_web_request" do
     before do
       current_time = Time.now