fix: reset subscribers on code reload in dev

Closes #10

CHANGELOG.md

@@ -2,6 +2,8 @@
 
 ## master
 
+- Reset subscribers on code reloading to avoid double execution.
+
 ## 1.5.0 (2024-04-26)
 
 - Require Ruby 2.7+.

lib/downstream/engine.rb

@@ -7,6 +7,7 @@ class Engine < ::Rails::Engine
     config.downstream = Downstream.config
 
     config.to_prepare do
+      Downstream.pubsub.reset
       ActiveSupport.run_load_hooks("downstream-events", Downstream)
     end
   end

lib/downstream/pubsub_adapters/abstract_pubsub.rb

@@ -2,6 +2,10 @@
 
 module Downstream
   class AbstractPubsub
+    def reset
+      raise NotImplementedError
+    end
+
     def subscribe(identifier, callable)
       raise NotImplementedError
     end

lib/downstream/pubsub_adapters/stateless/pubsub.rb

@@ -6,8 +6,20 @@
 module Downstream
   module Stateless
     class Pubsub < AbstractPubsub
+      def initialize
+        @subscribers = []
+      end
+
+      def reset
+        @subscribers.each(&:unsubscribe)
+        @subscribers.clear
+      end
+
       def subscribe(identifier, callable, async: false)
-        Subscriber.new(callable, async: async).tap { |s| s.subscribe(identifier) }
+        Subscriber.new(callable, async: async).tap do |s|
+          s.subscribe(identifier)
+          @subscribers << s
+        end
       end
 
       def subscribed(identifier, callable, &block)

spec/cases/code_reloading_spec.rb

@@ -0,0 +1,46 @@
+# frozen_string_literal: true
+
+require "spec_helper"
+
+describe "Rails #to_prepare" do
+  let!(:event_class) { Downstream::TestEvent }
+
+  let!(:callable) do
+    Downstream::TestSubscriber =
+      Module.new do
+        class << self
+          def events
+            @events ||= []
+          end
+
+          def call(event)
+            events << event
+          end
+        end
+      end
+  end
+
+  let(:event) { event_class.new(user_id: 0, action_type: :join) }
+  let(:event2) { event_class.new(user_id: 1, action_type: :leave) }
+
+  after do
+    Downstream.send(:remove_const, :TestSubscriber) if
+      Downstream.const_defined?(:TestSubscriber)
+  end
+
+  it "reset subscribers on #to_prepare" do
+    ActiveSupport.on_load "downstream-events" do
+      Downstream.subscribe(Downstream::TestSubscriber, to: Downstream::TestEvent)
+    end
+
+    Downstream.publish(event)
+
+    expect(callable.events.size).to eq 1
+
+    Rails.application.reloader.prepare!
+
+    Downstream.publish(event2)
+
+    expect(callable.events.size).to eq 2
+  end
+end