diff options
Diffstat (limited to 'actioncable/lib/action_cable/subscription_adapter')
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 |