aboutsummaryrefslogtreecommitdiffstats
path: root/actioncable/lib
diff options
context:
space:
mode:
Diffstat (limited to 'actioncable/lib')
-rw-r--r--actioncable/lib/action_cable.rb1
-rw-r--r--actioncable/lib/action_cable/channel/base.rb4
-rw-r--r--actioncable/lib/action_cable/channel/streams.rb6
-rw-r--r--actioncable/lib/action_cable/connection/base.rb2
-rw-r--r--actioncable/lib/action_cable/connection/internal_channel.rb12
-rw-r--r--actioncable/lib/action_cable/engine.rb6
-rw-r--r--actioncable/lib/action_cable/remote_connections.rb2
-rw-r--r--actioncable/lib/action_cable/server/base.rb17
-rw-r--r--actioncable/lib/action_cable/server/broadcasting.rb9
-rw-r--r--actioncable/lib/action_cable/server/configuration.rb24
-rw-r--r--actioncable/lib/action_cable/subscription_adapter.rb5
-rw-r--r--actioncable/lib/action_cable/subscription_adapter/base.rb24
-rw-r--r--actioncable/lib/action_cable/subscription_adapter/postgresql.rb98
-rw-r--r--actioncable/lib/action_cable/subscription_adapter/redis.rb37
14 files changed, 205 insertions, 42 deletions
diff --git a/actioncable/lib/action_cable.rb b/actioncable/lib/action_cable.rb
index 97f485b32e..1dc66ef3ad 100644
--- a/actioncable/lib/action_cable.rb
+++ b/actioncable/lib/action_cable.rb
@@ -47,4 +47,5 @@ module ActionCable
autoload :Connection
autoload :Channel
autoload :RemoteConnections
+ autoload :SubscriptionAdapter
end
diff --git a/actioncable/lib/action_cable/channel/base.rb b/actioncable/lib/action_cable/channel/base.rb
index ce9d62635c..88cdc1cab1 100644
--- a/actioncable/lib/action_cable/channel/base.rb
+++ b/actioncable/lib/action_cable/channel/base.rb
@@ -133,8 +133,8 @@ module ActionCable
@identifier = identifier
@params = params
- # When a channel is streaming via redis pubsub, we want to delay the confirmation
- # transmission until redis pubsub subscription is confirmed.
+ # When a channel is streaming via pubsub, we want to delay the confirmation
+ # transmission until pubsub subscription is confirmed.
@defer_subscription_confirmation = false
@reject_subscription = nil
diff --git a/actioncable/lib/action_cable/channel/streams.rb b/actioncable/lib/action_cable/channel/streams.rb
index b5ffa17f72..589946c3db 100644
--- a/actioncable/lib/action_cable/channel/streams.rb
+++ b/actioncable/lib/action_cable/channel/streams.rb
@@ -76,10 +76,10 @@ module ActionCable
streams << [ broadcasting, callback ]
EM.next_tick do
- pubsub.subscribe(broadcasting, &callback).callback do |reply|
+ pubsub.subscribe(broadcasting, callback, lambda do |reply|
transmit_subscription_confirmation
logger.info "#{self.class.name} is streaming from #{broadcasting}"
- end
+ end)
end
end
@@ -92,7 +92,7 @@ module ActionCable
def stop_all_streams
streams.each do |broadcasting, callback|
- pubsub.unsubscribe_proc broadcasting, callback
+ pubsub.unsubscribe broadcasting, callback
logger.info "#{self.class.name} stopped streaming from #{broadcasting}"
end.clear
end
diff --git a/actioncable/lib/action_cable/connection/base.rb b/actioncable/lib/action_cable/connection/base.rb
index a8cfdf90f3..bb8850aaa0 100644
--- a/actioncable/lib/action_cable/connection/base.rb
+++ b/actioncable/lib/action_cable/connection/base.rb
@@ -60,7 +60,7 @@ module ActionCable
@subscriptions = ActionCable::Connection::Subscriptions.new(self)
@message_buffer = ActionCable::Connection::MessageBuffer.new(self)
- @_internal_redis_subscriptions = nil
+ @_internal_subscriptions = nil
@started_at = Time.now
end
diff --git a/actioncable/lib/action_cable/connection/internal_channel.rb b/actioncable/lib/action_cable/connection/internal_channel.rb
index c065a24ab7..54ed7672d2 100644
--- a/actioncable/lib/action_cable/connection/internal_channel.rb
+++ b/actioncable/lib/action_cable/connection/internal_channel.rb
@@ -5,24 +5,24 @@ module ActionCable
extend ActiveSupport::Concern
private
- def internal_redis_channel
+ def internal_channel
"action_cable/#{connection_identifier}"
end
def subscribe_to_internal_channel
if connection_identifier.present?
callback = -> (message) { process_internal_message(message) }
- @_internal_redis_subscriptions ||= []
- @_internal_redis_subscriptions << [ internal_redis_channel, callback ]
+ @_internal_subscriptions ||= []
+ @_internal_subscriptions << [ internal_channel, callback ]
- EM.next_tick { pubsub.subscribe(internal_redis_channel, &callback) }
+ EM.next_tick { pubsub.subscribe(internal_channel, callback) }
logger.info "Registered connection (#{connection_identifier})"
end
end
def unsubscribe_from_internal_channel
- if @_internal_redis_subscriptions.present?
- @_internal_redis_subscriptions.each { |channel, callback| EM.next_tick { pubsub.unsubscribe_proc(channel, callback) } }
+ if @_internal_subscriptions.present?
+ @_internal_subscriptions.each { |channel, callback| EM.next_tick { pubsub.unsubscribe(channel, callback) } }
end
end
diff --git a/actioncable/lib/action_cable/engine.rb b/actioncable/lib/action_cable/engine.rb
index 2d3caa5b0a..f5e233e091 100644
--- a/actioncable/lib/action_cable/engine.rb
+++ b/actioncable/lib/action_cable/engine.rb
@@ -24,11 +24,11 @@ module ActionCable
options = app.config.action_cable
options.allowed_request_origins ||= "http://localhost:3000" if ::Rails.env.development?
- app.paths.add "config/redis/cable", with: "config/redis/cable.yml"
+ app.paths.add "config/cable", with: "config/cable.yml"
ActiveSupport.on_load(:action_cable) do
- if (redis_cable_path = Pathname.new(app.config.paths["config/redis/cable"].first)).exist?
- self.redis = Rails.application.config_for(redis_cable_path).with_indifferent_access
+ if (config_path = Pathname.new(app.config.paths["config/cable"].first)).exist?
+ self.cable = Rails.application.config_for(config_path).with_indifferent_access
end
options.each { |k,v| send("#{k}=", v) }
diff --git a/actioncable/lib/action_cable/remote_connections.rb b/actioncable/lib/action_cable/remote_connections.rb
index 1230d905ad..aa2fc95d2f 100644
--- a/actioncable/lib/action_cable/remote_connections.rb
+++ b/actioncable/lib/action_cable/remote_connections.rb
@@ -39,7 +39,7 @@ module ActionCable
# Uses the internal channel to disconnect the connection.
def disconnect
- server.broadcast internal_redis_channel, type: 'disconnect'
+ server.broadcast internal_channel, type: 'disconnect'
end
# Returns all the identifiers that were applied to this connection.
diff --git a/actioncable/lib/action_cable/server/base.rb b/actioncable/lib/action_cable/server/base.rb
index 3785bbd154..3385a4c9f3 100644
--- a/actioncable/lib/action_cable/server/base.rb
+++ b/actioncable/lib/action_cable/server/base.rb
@@ -1,5 +1,3 @@
-require 'em-hiredis'
-
module ActionCable
module Server
# A singleton ActionCable::Server instance is available via ActionCable.server. It's used by the rack process that starts the cable server, but
@@ -47,20 +45,9 @@ module ActionCable
end
end
- # The redis pubsub adapter used for all streams/broadcasting.
+ # Adapter used for all streams/broadcasting.
def pubsub
- @pubsub ||= redis.pubsub
- end
-
- # The EventMachine Redis instance used by the pubsub adapter.
- def redis
- @redis ||= EM::Hiredis.connect(config.redis[:url]).tap do |redis|
- redis.on(:reconnect_failed) do
- logger.info "[ActionCable] Redis reconnect failed."
- # logger.info "[ActionCable] Redis reconnected. Closing all the open connections."
- # @connections.map &:close
- end
- end
+ @pubsub ||= config.pubsub_adapter.new(self)
end
# All the identifiers applied to the connection class associated with this server.
diff --git a/actioncable/lib/action_cable/server/broadcasting.rb b/actioncable/lib/action_cable/server/broadcasting.rb
index c759239a0e..4a26ed9269 100644
--- a/actioncable/lib/action_cable/server/broadcasting.rb
+++ b/actioncable/lib/action_cable/server/broadcasting.rb
@@ -1,5 +1,3 @@
-require 'redis'
-
module ActionCable
module Server
# Broadcasting is how other parts of your application can send messages to the channel subscribers. As explained in Channel, most of the time, these
@@ -31,11 +29,6 @@ module ActionCable
Broadcaster.new(self, broadcasting)
end
- # The redis instance used for broadcasting. Not intended for direct user use.
- def broadcasting_redis
- @broadcasting_redis ||= Redis.new(config.redis)
- end
-
private
class Broadcaster
attr_reader :server, :broadcasting
@@ -46,7 +39,7 @@ module ActionCable
def broadcast(message)
server.logger.info "[ActionCable] Broadcasting to #{broadcasting}: #{message}"
- server.broadcasting_redis.publish broadcasting, ActiveSupport::JSON.encode(message)
+ server.pubsub.broadcast broadcasting, ActiveSupport::JSON.encode(message)
end
end
end
diff --git a/actioncable/lib/action_cable/server/configuration.rb b/actioncable/lib/action_cable/server/configuration.rb
index 935133cbba..ebbf60c6e2 100644
--- a/actioncable/lib/action_cable/server/configuration.rb
+++ b/actioncable/lib/action_cable/server/configuration.rb
@@ -5,9 +5,9 @@ module ActionCable
class Configuration
attr_accessor :logger, :log_tags
attr_accessor :connection_class, :worker_pool_size
- attr_accessor :redis, :channels_path
+ attr_accessor :channels_path
attr_accessor :disable_request_forgery_protection, :allowed_request_origins
- attr_accessor :url
+ attr_accessor :cable, :url
def initialize
@log_tags = []
@@ -29,7 +29,25 @@ module ActionCable
Pathname.new(channel_path).basename.to_s.split('.').first.camelize
end
end
+
+ # Returns constant of subscription adapter specified in config/cable.yml.
+ # If the adapter cannot be found, this will default to the Redis adapter.
+ # Also makes sure proper dependencies are required.
+ def pubsub_adapter
+ adapter = (cable.fetch('adapter') { 'redis' })
+ path_to_adapter = "action_cable/subscription_adapter/#{adapter}"
+ begin
+ require path_to_adapter
+ rescue Gem::LoadError => e
+ raise Gem::LoadError, "Specified '#{adapter}' for Action Cable pubsub adapter, but the gem is not loaded. Add `gem '#{e.name}'` to your Gemfile (and ensure its version is at the minimum required by Action Cable)."
+ rescue LoadError => e
+ raise LoadError, "Could not load '#{path_to_adapter}'. Make sure that the adapter in config/cable.yml is valid. If you use an adapter other than 'postgresql' or 'redis' add the necessary adapter gem to the Gemfile.", e.backtrace
+ end
+
+ adapter = adapter.camelize
+ adapter = 'PostgreSQL' if adapter == 'Postgresql'
+ "ActionCable::SubscriptionAdapter::#{adapter}".constantize
+ end
end
end
end
-
diff --git a/actioncable/lib/action_cable/subscription_adapter.rb b/actioncable/lib/action_cable/subscription_adapter.rb
new file mode 100644
index 0000000000..e770f4fb00
--- /dev/null
+++ b/actioncable/lib/action_cable/subscription_adapter.rb
@@ -0,0 +1,5 @@
+module ActionCable
+ module SubscriptionAdapter
+ autoload :Base, 'action_cable/subscription_adapter/base'
+ end
+end
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
generated by cgit v1.2.3 (git 2.25.1) at 2024-07-03 14:27:40 +0000