Merge pull request #22950 from maclover7/adapterize-storage-actioncable

Adapterize storage for ActionCable

3 files changed, 159 insertions, 0 deletions

diff --git a/actioncable/lib/action_cable/subscription_adapter/base.rb b/actioncable/lib/action_cable/subscription_adapter/base.rb
new file mode 100644
index 0000000000..11910803e8
--- /dev/null
+++ b/actioncable/lib/action_cable/subscription_adapter/base.rb
@@ -0,0 +1,24 @@
+module ActionCable
+  module SubscriptionAdapter
+    class Base
+      attr_reader :logger, :server
+
+      def initialize(server)
+        @server = server
+        @logger = @server.logger
+      end
+
+      def broadcast(channel, payload)
+        raise NotImplementedError
+      end
+
+      def subscribe(channel, message_callback, success_callback = nil)
+        raise NotImplementedError
+      end
+
+      def unsubscribe(channel, message_callback)
+        raise NotImplementedError
+      end
+    end
+  end
+end
diff --git a/actioncable/lib/action_cable/subscription_adapter/postgresql.rb b/actioncable/lib/action_cable/subscription_adapter/postgresql.rb
new file mode 100644
index 0000000000..6465663c97
--- /dev/null
+++ b/actioncable/lib/action_cable/subscription_adapter/postgresql.rb
@@ -0,0 +1,98 @@
+gem 'pg', '~> 0.18'
+require 'pg'
+require 'thread'
+
+module ActionCable
+  module SubscriptionAdapter
+    class PostgreSQL < Base # :nodoc:
+      def broadcast(channel, payload)
+        with_connection do |pg_conn|
+          pg_conn.exec("NOTIFY #{pg_conn.escape_identifier(channel)}, '#{pg_conn.escape_string(payload)}'")
+        end
+      end
+
+      def subscribe(channel, callback, success_callback = nil)
+        listener.subscribe_to(channel, callback, success_callback)
+      end
+
+      def unsubscribe(channel, callback)
+        listener.unsubscribe_from(channel, callback)
+      end
+
+      def with_connection(&block) # :nodoc:
+        ActiveRecord::Base.connection_pool.with_connection do |ar_conn|
+          pg_conn = ar_conn.raw_connection
+
+          unless pg_conn.is_a?(PG::Connection)
+            raise 'ActiveRecord database must be Postgres in order to use the Postgres ActionCable storage adapter'
+          end
+
+          yield pg_conn
+        end
+      end
+
+      private
+        def listener
+          @listener ||= Listener.new(self)
+        end
+
+        class Listener
+          def initialize(adapter)
+            @adapter = adapter
+            @subscribers = Hash.new { |h,k| h[k] = [] }
+            @sync = Mutex.new
+            @queue = Queue.new
+
+            Thread.new do
+              Thread.current.abort_on_exception = true
+              listen
+            end
+          end
+
+          def listen
+            @adapter.with_connection do |pg_conn|
+              loop do
+                until @queue.empty?
+                  action, channel, callback = @queue.pop(true)
+                  escaped_channel = pg_conn.escape_identifier(channel)
+
+                  if action == :listen
+                    pg_conn.exec("LISTEN #{escaped_channel}")
+                    ::EM.next_tick(&callback) if callback
+                  elsif action == :unlisten
+                    pg_conn.exec("UNLISTEN #{escaped_channel}")
+                  end
+                end
+
+                pg_conn.wait_for_notify(1) do |chan, pid, message|
+                  @subscribers[chan].each do |callback|
+                    ::EM.next_tick { callback.call(message) }
+                  end
+                end
+              end
+            end
+          end
+
+          def subscribe_to(channel, callback, success_callback)
+            @sync.synchronize do
+              if @subscribers[channel].empty?
+                @queue.push([:listen, channel, success_callback])
+              end
+
+              @subscribers[channel] << callback
+            end
+          end
+
+          def unsubscribe_from(channel, callback)
+            @sync.synchronize do
+              @subscribers[channel].delete(callback)
+
+              if @subscribers[channel].empty?
+                @queue.push([:unlisten, channel])
+              end
+            end
+          end
+        end
+    end
+  end
+end
diff --git a/actioncable/lib/action_cable/subscription_adapter/redis.rb b/actioncable/lib/action_cable/subscription_adapter/redis.rb
new file mode 100644
index 0000000000..d149f28b1f
--- /dev/null
+++ b/actioncable/lib/action_cable/subscription_adapter/redis.rb
@@ -0,0 +1,37 @@
+gem 'em-hiredis', '~> 0.3.0'
+gem 'redis', '~> 3.0'
+require 'em-hiredis'
+require 'redis'
+
+module ActionCable
+  module SubscriptionAdapter
+    class Redis < Base # :nodoc:
+      def broadcast(channel, payload)
+        redis_connection_for_broadcasts.publish(channel, payload)
+      end
+
+      def subscribe(channel, message_callback, success_callback = nil)
+        redis_connection_for_subscriptions.pubsub.subscribe(channel, &message_callback).tap do |result|
+          result.callback(&success_callback) if success_callback
+        end
+      end
+
+      def unsubscribe(channel, message_callback)
+        hi_redis_conn.pubsub.unsubscribe_proc(channel, message_callback)
+      end
+
+      private
+        def redis_connection_for_subscriptions
+          @redis_connection_for_subscriptions ||= EM::Hiredis.connect(@server.config.cable[:url]).tap do |redis|
+            redis.on(:reconnect_failed) do
+              @logger.info "[ActionCable] Redis reconnect failed."
+            end
+          end
+        end
+
+        def redis_connection_for_broadcasts
+          @redis_connection_for_broadcasts ||= ::Redis.new(@server.config.cable)
+        end
+    end
+  end
+end