Prevent broadcasting old messages when new channels are subscribed (#67)

Author: npezza93

app/models/solid_cable/message.rb

@@ -7,7 +7,7 @@ class Message < SolidCable::Record
     }
     scope :broadcastable, lambda { |channels, last_id|
       where(channel_hash: channel_hashes_for(channels)).
-        where(id: (last_id + 1)..).order(:id)
+        where(id: (last_id.to_i + 1)..).order(:id)
     }
 
     class << self

lib/action_cable/subscription_adapter/solid_cable.rb

@@ -89,7 +89,7 @@ def shutdown
           end
 
           def add_channel(channel, on_success)
-            channels.add(channel)
+            channels[channel] = last_message_id
             event_loop.post(&on_success) if on_success
           end
 
@@ -103,21 +103,22 @@ def invoke_callback(*)
 
           private
             attr_reader :event_loop, :thread
-            attr_writer :last_id
 
-            def last_id
-              @last_id ||= ::SolidCable::Message.maximum(:id) || 0
+            def last_message_id
+              ::SolidCable::Message.maximum(:id) || 0
             end
 
             def channels
-              @channels ||= Set.new
+              @channels ||= Concurrent::Hash.new
             end
 
             def broadcast_messages
-              ::SolidCable::Message.broadcastable(channels, last_id).
+              ::SolidCable::Message.broadcastable(channels.keys, channels.values.min).
                 each do |message|
-                broadcast(message.channel, message.payload)
-                self.last_id = message.id
+                if channels[message.channel].present? && channels[message.channel] < message.id
+                  broadcast(message.channel, message.payload)
+                  channels[message.channel] = message.id
+                end
               end
             end
 

test/lib/action_cable/subscription_adapter/solid_cable_test.rb

@@ -148,6 +148,33 @@ class ActionCable::SubscriptionAdapter::SolidCableTest < ActionCable::TestCase
     end
   end
 
+  test "does not send old messages" do
+    @tx_adapter.broadcast("channel", "channel1")
+    @tx_adapter.broadcast("channel", "channel2")
+
+    subscribe_as_queue("channel") do |queue|
+      assert_empty queue
+
+      @tx_adapter.broadcast("channel", "channel3")
+      @tx_adapter.broadcast("channel", "channel4")
+      @tx_adapter.broadcast("other", "other1")
+      @tx_adapter.broadcast("other", "other2")
+
+      subscribe_as_queue("other") do |other_queue|
+        assert_empty other_queue
+      end
+      assert_equal "channel3", queue.pop
+      assert_equal "channel4", queue.pop
+    end
+
+    @tx_adapter.broadcast("channel", "channel5")
+    @tx_adapter.broadcast("channel", "channel6")
+
+    subscribe_as_queue("channel") do |queue|
+      assert_empty queue
+    end
+  end
+
   private
     def cable_config
       { adapter: "solid_cable", message_retention: "1.second",