Adds a helper for declaring heartbeat subscriptions

At TaskRabbit, we use the heartbeat to schedule regular work. This has
resulted in a fairly fragmented implementation of the modulo and
matching logic. By wrapping this in a helper function we can more easily
search and grep for these subscriptions.

Author: choubacha

CHANGELOG.md

@@ -6,6 +6,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+### Added
+
+- Adds `Dispatch#on_heartbeatch` which is a helper function for specifying heartbeat subscriptions.
+
 ## [0.12.0]
 
 ### Changed

README.mdown

@@ -63,6 +63,16 @@ QueueBus.dispatch("app_b") do
   subscribe "my_key", { "user_id" => :present, "page" => "homepage"} do
     Mixpanel.homepage_action!(attributes["action"])
   end
+
+  # You may also declare a subscription to heartbeat events. This is a helper function
+  # that works along with subscribe to make scheduling regular events easier.
+  #
+  #   minute_interval: Executes every n minutes
+  #   hour_interval:   Executes every n hours
+  #   minute:          Executes on this minute
+  #   hour:            Executes on this hour
+  on_heartbeat "my_heartbeat_event", minute_interval: 5 do |attributes|
+  end
 end
 ```
 

lib/queue_bus/dispatch.rb

@@ -16,6 +16,39 @@ def size
       @subscriptions.size
     end
 
+    def on_heartbeat(key, minute: nil, hour: nil, minute_interval: nil, hour_interval: nil, &block) # rubocop:disable Metrics/PerceivedComplexity, Metrics/MethodLength, Metrics/ParameterLists, Metrics/CyclomaticComplexity, Metrics/AbcSize
+      if minute_interval && !minute_interval.positive?
+        raise ArgumentError, 'minute_interval must be a positive integer'
+      end
+
+      if hour_interval && !hour_interval.positive?
+        raise ArgumentError, 'hour_interval must be a positive integer'
+      end
+
+      matcher = { bus_event_type: :heartbeat_minutes }
+
+      if minute
+        raise ArgumentError, 'minute must be a positive integer' unless minute.positive?
+
+        matcher['minute'] = minute
+      end
+
+      if hour
+        raise ArgumentError, 'hour must be a positive integer' unless hour.positive?
+
+        matcher['hour'] = hour
+      end
+
+      subscribe(key, matcher) do |event|
+        if (minute_interval.nil? || (event['minute'] % minute_interval).zero?) &&
+           (hour_interval.nil? || (event['hour'] % hour_interval).zero?)
+
+          # Yield the block passed in.
+          block.call
+        end
+      end
+    end
+
     def subscribe(key, matcher_hash = nil, &block)
       dispatch_event('default', key, matcher_hash, block)
     end

spec/dispatch_spec.rb

@@ -28,6 +28,137 @@ module QueueBus
       end.not_to raise_error
     end
 
+    describe '#on_heartbeat' do
+      let(:dispatch) { Dispatch.new('heartbeat') }
+      let(:event) { { bus_event_type: :heartbeat_minutes } }
+      let(:event_name) { 'my-event' }
+
+      context 'when not declaring anything' do
+        before do
+          dispatch.on_heartbeat event_name do |_event|
+            Runner2.run({})
+          end
+        end
+
+        it 'runs on every heart beat' do
+          (0..24).each do |hour|
+            (0..60).each do |minute|
+              expect do
+                dispatch.execute(
+                  event_name, event.merge('hour' => hour, 'minute' => minute)
+                )
+              end.to change(Runner2, :value).by(1)
+            end
+          end
+        end
+      end
+
+      context 'when running on hour 8' do
+        before do
+          dispatch.on_heartbeat event_name, hour: 8 do |_event|
+            Runner2.run({})
+          end
+        end
+
+        it 'subscribes to hour 8' do
+          expect(dispatch.subscriptions.all.first.matcher.filters).to eq('bus_event_type' => 'heartbeat_minutes', 'hour' => '8')
+        end
+      end
+
+      context 'when running on minute 4' do
+        before do
+          dispatch.on_heartbeat event_name, minute: 4 do |_event|
+            Runner2.run({})
+          end
+        end
+
+        it 'subscribes to minute 4' do
+          expect(dispatch.subscriptions.all.first.matcher.filters).to eq('bus_event_type' => 'heartbeat_minutes', 'minute' => '4')
+        end
+      end
+
+      context 'when running on minute 4 and hour 8' do
+        before do
+          dispatch.on_heartbeat event_name, hour: 8, minute: 4 do |_event|
+            Runner2.run({})
+          end
+        end
+
+        it 'subscribes to minute 4 and hour 8' do
+          expect(dispatch.subscriptions.all.first.matcher.filters)
+            .to eq('bus_event_type' => 'heartbeat_minutes', 'minute' => '4', 'hour' => '8')
+        end
+      end
+
+      context 'when declaring minute intervals' do
+        before do
+          dispatch.on_heartbeat event_name, minute_interval: 5 do |_event|
+            Runner2.run({})
+          end
+        end
+
+        it 'runs the runner when the minute buzzes (modulos to 5)' do
+          (0..60).each do |minute|
+            if minute % 5 == 0
+              expect { dispatch.execute(event_name, event.merge('minute' => minute)) }
+                .to change(Runner2, :value).by(1)
+            else
+              expect { dispatch.execute(event_name, event.merge('minute' => minute)) }
+                .not_to change(Runner2, :value)
+            end
+          end
+        end
+      end
+
+      context 'when declaring hour intervals' do
+        before do
+          dispatch.on_heartbeat event_name, hour_interval: 3 do |_event|
+            Runner2.run({})
+          end
+        end
+
+        it 'runs the runner when the hour fizzes (modulos to 3)' do
+          (0..60).each do |hour|
+            if hour % 3 == 0
+              expect { dispatch.execute(event_name, event.merge('hour' => hour)) }
+                .to change(Runner2, :value).by(1)
+            else
+              expect { dispatch.execute(event_name, event.merge('hour' => hour)) }
+                .not_to change(Runner2, :value)
+            end
+          end
+        end
+      end
+
+      context 'when declaring hour and minute intervals' do
+        before do
+          dispatch.on_heartbeat event_name, minute_interval: 5, hour_interval: 3 do |_event|
+            Runner2.run({})
+          end
+        end
+
+        it 'runs the runner when the time fizzbuzzes (modulos to 3 and 5)' do
+          (0..24).each do |hour|
+            (0..60).each do |minute|
+              if hour % 3 == 0 && minute % 5 == 0
+                expect do
+                  dispatch.execute(
+                    event_name, event.merge('hour' => hour, 'minute' => minute)
+                  )
+                end.to change(Runner2, :value).by(1)
+              else
+                expect do
+                  dispatch.execute(
+                    event_name, event.merge('hour' => hour, 'minute' => minute)
+                  )
+                end.not_to change(Runner2, :value)
+              end
+            end
+          end
+        end
+      end
+    end
+
     describe 'Top Level' do
       before(:each) do
         QueueBus.dispatch('testit') do