MessageBus handles readonly redis now, no need to wrap it

This commit is contained in:
Sam 2015-05-04 12:21:00 +10:00
parent 1c7fa956dc
commit 803feefd54
29 changed files with 111 additions and 140 deletions

View File

@ -54,7 +54,7 @@ class Admin::UsersController < Admin::AdminController
@user.suspended_at = DateTime.now
@user.save!
StaffActionLogger.new(current_user).log_user_suspend(@user, params[:reason])
DiscourseBus.publish "/logout", @user.id, user_ids: [@user.id]
MessageBus.publish "/logout", @user.id, user_ids: [@user.id]
render nothing: true
end
@ -71,7 +71,7 @@ class Admin::UsersController < Admin::AdminController
if @user
@user.auth_token = nil
@user.save!
DiscourseBus.publish "/logout", @user.id, user_ids: [@user.id]
MessageBus.publish "/logout", @user.id, user_ids: [@user.id]
render json: success_json
else
render json: {error: I18n.t('admin_js.admin.users.id_not_found')}, status: 404
@ -350,7 +350,7 @@ class Admin::UsersController < Admin::AdminController
end
def refresh_browser(user)
DiscourseBus.publish "/file-change", ["refresh"], user_ids: [user.id]
MessageBus.publish "/file-change", ["refresh"], user_ids: [user.id]
end
end

View File

@ -222,7 +222,7 @@ SQL
end
def publish_categories_list
DiscourseBus.publish('/categories', {categories: ActiveModel::ArraySerializer.new(Category.latest).as_json})
MessageBus.publish('/categories', {categories: ActiveModel::ArraySerializer.new(Category.latest).as_json})
end
def parent_category_validator

View File

@ -97,7 +97,7 @@ class ColorScheme < ActiveRecord::Base
end
def publish_discourse_stylesheet
DiscourseBus.publish("/discourse_stylesheet", self.name)
MessageBus.publish("/discourse_stylesheet", self.name)
DiscourseStylesheets.cache.clear
end

View File

@ -97,7 +97,7 @@ class Post < ActiveRecord::Base
# special failsafe for posts missing topics
# consistency checks should fix, but message
# is safe to skip
DiscourseBus.publish("/topic/#{topic_id}", {
MessageBus.publish("/topic/#{topic_id}", {
id: id,
post_number: post_number,
updated_at: Time.now,

View File

@ -56,7 +56,7 @@ class PostAction < ActiveRecord::Base
$redis.set('posts_flagged_count', posts_flagged_count)
user_ids = User.staff.pluck(:id)
DiscourseBus.publish('/flagged_counts', { total: posts_flagged_count }, { user_ids: user_ids })
MessageBus.publish('/flagged_counts', { total: posts_flagged_count }, { user_ids: user_ids })
end
def self.flagged_posts_count

View File

@ -42,7 +42,7 @@ class QueuedPost < ActiveRecord::Base
def self.broadcast_new!
msg = { post_queue_new_count: QueuedPost.new_count }
DiscourseBus.publish('/queue_counts', msg, user_ids: User.staff.pluck(:id))
MessageBus.publish('/queue_counts', msg, user_ids: User.staff.pluck(:id))
end
def reject!(rejected_by)

View File

@ -34,11 +34,11 @@ class SiteCustomization < ActiveRecord::Base
after_save do
remove_from_cache!
if stylesheet_changed? || mobile_stylesheet_changed?
DiscourseBus.publish "/file-change/#{key}", SecureRandom.hex
DiscourseBus.publish "/file-change/#{SiteCustomization::ENABLED_KEY}", SecureRandom.hex
MessageBus.publish "/file-change/#{key}", SecureRandom.hex
MessageBus.publish "/file-change/#{SiteCustomization::ENABLED_KEY}", SecureRandom.hex
end
DiscourseBus.publish "/header-change/#{key}", header if header_changed?
DiscourseBus.publish "/footer-change/#{key}", footer if footer_changed?
MessageBus.publish "/header-change/#{key}", header if header_changed?
MessageBus.publish "/footer-change/#{key}", footer if footer_changed?
DiscourseStylesheets.cache.clear
end
@ -109,7 +109,7 @@ class SiteCustomization < ActiveRecord::Base
end
def self.remove_from_cache!(key, broadcast = true)
DiscourseBus.publish('/site_customization', key: key) if broadcast
MessageBus.publish('/site_customization', key: key) if broadcast
clear_cache!
end

View File

@ -641,7 +641,7 @@ class Topic < ActiveRecord::Base
self.add_moderator_post(user, I18n.t("archetypes.banner.message.make"))
self.save
DiscourseBus.publish('/site/banner', banner)
MessageBus.publish('/site/banner', banner)
end
def remove_banner!(user)
@ -649,7 +649,7 @@ class Topic < ActiveRecord::Base
self.add_moderator_post(user, I18n.t("archetypes.banner.message.remove"))
self.save
DiscourseBus.publish('/site/banner', nil)
MessageBus.publish('/site/banner', nil)
end
def banner

View File

@ -33,7 +33,7 @@ class TopicTrackingState
group_ids = topic.category && topic.category.secure_group_ids
DiscourseBus.publish("/new", message.as_json, group_ids: group_ids)
MessageBus.publish("/new", message.as_json, group_ids: group_ids)
publish_read(topic.id, 1, topic.user_id)
end
@ -51,7 +51,7 @@ class TopicTrackingState
}
group_ids = topic.category && topic.category.secure_group_ids
DiscourseBus.publish("/latest", message.as_json, group_ids: group_ids)
MessageBus.publish("/latest", message.as_json, group_ids: group_ids)
end
def self.publish_unread(post)
@ -77,7 +77,7 @@ class TopicTrackingState
}
}
DiscourseBus.publish("/unread/#{tu.user_id}", message.as_json, group_ids: group_ids)
MessageBus.publish("/unread/#{tu.user_id}", message.as_json, group_ids: group_ids)
end
end
@ -97,7 +97,7 @@ class TopicTrackingState
}
}
DiscourseBus.publish("/unread/#{user_id}", message.as_json, user_ids: [user_id])
MessageBus.publish("/unread/#{user_id}", message.as_json, user_ids: [user_id])
end

View File

@ -40,7 +40,7 @@ class TopicUser < ActiveRecord::Base
notifications_reason_id: reason
)
DiscourseBus.publish("/topic/#{topic_id}", {
MessageBus.publish("/topic/#{topic_id}", {
notification_level_change: notification_levels[:tracking],
notifications_reason_id: reason
}, user_ids: [user_id])
@ -113,7 +113,7 @@ class TopicUser < ActiveRecord::Base
end
if attrs[:notification_level]
DiscourseBus.publish("/topic/#{topic_id}",
MessageBus.publish("/topic/#{topic_id}",
{notification_level_change: attrs[:notification_level]}, user_ids: [user_id])
end
@ -196,7 +196,7 @@ class TopicUser < ActiveRecord::Base
end
if before != after
DiscourseBus.publish("/topic/#{topic_id}", {notification_level_change: after}, user_ids: [user.id])
MessageBus.publish("/topic/#{topic_id}", {notification_level_change: after}, user_ids: [user.id])
end
end
@ -219,7 +219,7 @@ class TopicUser < ActiveRecord::Base
WHERE ftu.user_id = :user_id and ftu.topic_id = :topic_id)",
args)
DiscourseBus.publish("/topic/#{topic_id}", {notification_level_change: args[:new_status]}, user_ids: [user.id])
MessageBus.publish("/topic/#{topic_id}", {notification_level_change: args[:new_status]}, user_ids: [user.id])
end
end

View File

@ -206,7 +206,7 @@ class User < ActiveRecord::Base
# tricky, we need our bus to be subscribed from the right spot
def sync_notification_channel_position
@unread_notifications_by_type = nil
self.notification_channel_position = DiscourseBus.last_id("/notification/#{id}")
self.notification_channel_position = MessageBus.last_id("/notification/#{id}")
end
def invited_by
@ -298,7 +298,7 @@ class User < ActiveRecord::Base
end
def publish_notifications_state
DiscourseBus.publish("/notification/#{id}",
MessageBus.publish("/notification/#{id}",
{unread_notifications: unread_notifications,
unread_private_messages: unread_private_messages,
total_unread_notifications: total_unread_notifications},

View File

@ -224,7 +224,7 @@ SQL
end
if action.user
DiscourseBus.publish("/users/#{action.user.username.downcase}", action.id, user_ids: [user_id], group_ids: group_ids)
MessageBus.publish("/users/#{action.user.username.downcase}", action.id, user_ids: [user_id], group_ids: group_ids)
end
action
@ -240,7 +240,7 @@ SQL
require_parameters(hash, :action_type, :user_id, :acting_user_id, :target_topic_id, :target_post_id)
if action = UserAction.find_by(hash.except(:created_at))
action.destroy
DiscourseBus.publish("/user/#{hash[:user_id]}", {user_action_id: action.id, remove: true})
MessageBus.publish("/user/#{hash[:user_id]}", {user_action_id: action.id, remove: true})
end
update_like_count(hash[:user_id], hash[:action_type], -1)

View File

@ -80,7 +80,7 @@ class UserDestroyer
end
StaffActionLogger.new(@actor == user ? Discourse.system_user : @actor).log_user_deletion(user, opts.slice(:context))
DiscourseBus.publish "/file-change", ["refresh"], user_ids: [user.id]
MessageBus.publish "/file-change", ["refresh"], user_ids: [user.id]
end
end
end

View File

@ -1,3 +0,0 @@
require_dependency 'discourse_bus'
DiscourseBus.start!

View File

@ -0,0 +1,47 @@
MessageBus.site_id_lookup do
RailsMultisite::ConnectionManagement.current_db
end
MessageBus.extra_response_headers_lookup do |env|
{
"Access-Control-Allow-Origin" => Discourse.base_url_no_prefix,
"Access-Control-Allow-Methods" => "GET, POST",
"Access-Control-Allow-Headers" => "X-SILENCE-LOGGER, X-Shared-Session-Key"
}
end
MessageBus.user_id_lookup do |env|
user = CurrentUser.lookup_from_env(env)
user.id if user
end
MessageBus.group_ids_lookup do |env|
user = CurrentUser.lookup_from_env(env)
user.groups.select('groups.id').map{|g| g.id} if user
end
MessageBus.on_connect do |site_id|
RailsMultisite::ConnectionManagement.establish_connection(db: site_id)
end
MessageBus.on_disconnect do |site_id|
ActiveRecord::Base.connection_handler.clear_active_connections!
end
# Point at our redis
MessageBus.redis_config = YAML.load(ERB.new(File.new("#{Rails.root}/config/redis.yml").read).result)[Rails.env].symbolize_keys
MessageBus.long_polling_enabled = SiteSetting.enable_long_polling
MessageBus.long_polling_interval = SiteSetting.long_polling_interval
MessageBus.is_admin_lookup do |env|
user = CurrentUser.lookup_from_env(env)
if user && user.admin
true
else
false
end
end
MessageBus.cache_assets = !Rails.env.development?
MessageBus.enable_diagnostics

View File

@ -85,7 +85,7 @@ class Auth::DefaultCurrentUserProvider
if SiteSetting.log_out_strict && (user = current_user)
user.auth_token = nil
user.save!
DiscourseBus.publish "/logout", user.id, user_ids: [user.id]
MessageBus.publish "/logout", user.id, user_ids: [user.id]
end
cookies[TOKEN_COOKIE] = nil
end

View File

@ -62,7 +62,7 @@ module BackupRestore
def self.logs
id = start_logs_message_id
DiscourseBus.backlog(LOGS_CHANNEL, id).map { |m| m.data }
MessageBus.backlog(LOGS_CHANNEL, id).map { |m| m.data }
end
def self.current_version
@ -142,7 +142,7 @@ module BackupRestore
end
def self.save_start_logs_message_id
id = DiscourseBus.last_id(LOGS_CHANNEL)
id = MessageBus.last_id(LOGS_CHANNEL)
$redis.set(start_logs_message_id_key, id)
end

View File

@ -336,7 +336,7 @@ module BackupRestore
def publish_log(message, timestamp)
return unless @publish_to_message_bus
data = { timestamp: timestamp, operation: "backup", message: message }
DiscourseBus.publish(BackupRestore::LOGS_CHANNEL, data, user_ids: [@user_id])
MessageBus.publish(BackupRestore::LOGS_CHANNEL, data, user_ids: [@user_id])
end
def save_log(message, timestamp)

View File

@ -354,7 +354,7 @@ module BackupRestore
def publish_log(message, timestamp)
return unless @publish_to_message_bus
data = { timestamp: timestamp, operation: "restore", message: message }
DiscourseBus.publish(BackupRestore::LOGS_CHANNEL, data, user_ids: [@user_id])
MessageBus.publish(BackupRestore::LOGS_CHANNEL, data, user_ids: [@user_id])
end
def save_log(message, timestamp)

View File

@ -119,10 +119,10 @@ module Discourse
digest = Digest::MD5.hexdigest(ActionView::Base.assets_manifest.assets.values.sort.join)
channel = "/global/asset-version"
message = DiscourseBus.last_message(channel)
message = MessageBus.last_message(channel)
unless message && message.data == digest
DiscourseBus.publish channel, digest
MessageBus.publish channel, digest
end
digest
end
@ -193,7 +193,7 @@ module Discourse
def self.enable_readonly_mode
$redis.set(readonly_mode_key, 1)
DiscourseBus.publish(readonly_channel, true)
MessageBus.publish(readonly_channel, true)
keep_readonly_mode
true
end
@ -210,7 +210,7 @@ module Discourse
def self.disable_readonly_mode
$redis.del(readonly_mode_key)
DiscourseBus.publish(readonly_channel, false)
MessageBus.publish(readonly_channel, false)
true
end
@ -221,9 +221,9 @@ module Discourse
def self.request_refresh!
# Causes refresh on next click for all clients
#
# This is better than `DiscourseBus.publish "/file-change", ["refresh"]` because
# This is better than `MessageBus.publish "/file-change", ["refresh"]` because
# it spreads the refreshes out over a time period
DiscourseBus.publish '/global/asset-version', 'clobber'
MessageBus.publish '/global/asset-version', 'clobber'
end
def self.git_version
@ -298,7 +298,7 @@ module Discourse
def self.after_fork
current_db = RailsMultisite::ConnectionManagement.current_db
RailsMultisite::ConnectionManagement.establish_connection(db: current_db)
DiscourseBus.after_fork
MessageBus.after_fork
SiteSetting.after_fork
$redis.client.reconnect
Rails.cache.reconnect

View File

@ -1,73 +0,0 @@
# Proxies MessageBus and noops READONLY errors.
module DiscourseBus
def self.start!
MessageBus.site_id_lookup do
RailsMultisite::ConnectionManagement.current_db
end
MessageBus.extra_response_headers_lookup do |env|
{
"Access-Control-Allow-Origin" => Discourse.base_url_no_prefix,
"Access-Control-Allow-Methods" => "GET, POST",
"Access-Control-Allow-Headers" => "X-SILENCE-LOGGER, X-Shared-Session-Key"
}
end
MessageBus.user_id_lookup do |env|
user = CurrentUser.lookup_from_env(env)
user.id if user
end
MessageBus.group_ids_lookup do |env|
user = CurrentUser.lookup_from_env(env)
user.groups.select('groups.id').map{|g| g.id} if user
end
MessageBus.on_connect do |site_id|
RailsMultisite::ConnectionManagement.establish_connection(db: site_id)
end
MessageBus.on_disconnect do |site_id|
ActiveRecord::Base.connection_handler.clear_active_connections!
end
# Point at our redis
MessageBus.redis_config = YAML.load(ERB.new(File.new("#{Rails.root}/config/redis.yml").read).result)[Rails.env].symbolize_keys
MessageBus.long_polling_enabled = SiteSetting.enable_long_polling
MessageBus.long_polling_interval = SiteSetting.long_polling_interval
MessageBus.is_admin_lookup do |env|
user = CurrentUser.lookup_from_env(env)
if user && user.admin
true
else
false
end
end
MessageBus.cache_assets = !Rails.env.development?
MessageBus.enable_diagnostics
end
def self.proxy_readonly(*methods)
methods.each do |name|
define_singleton_method(name) do |*args, &block|
DiscourseRedis.ignore_readonly { MessageBus.send(name, *args, &block) }
end
end
end
proxy_readonly :publish,
:subscribe,
:unsubscribe,
:last_id,
:last_message,
:backlog,
:after_fork,
:on_connect,
:on_disconnect,
:track_publish
end

View File

@ -51,7 +51,7 @@ class DistributedCache
return if @subscribed
@lock.synchronize do
return if @subscribed
DiscourseBus.subscribe(channel_name) do |message|
MessageBus.subscribe(channel_name) do |message|
@lock.synchronize do
process_message(message)
end
@ -63,7 +63,7 @@ class DistributedCache
def self.publish(hash, message)
message[:origin] = hash.identity
message[:hash_key] = hash.key
DiscourseBus.publish(channel_name, message, { user_ids: [-1] })
MessageBus.publish(channel_name, message, { user_ids: [-1] })
end
def self.set(hash, key, value)

View File

@ -11,7 +11,7 @@ class MessageBusDiags
end
def self.establish_peer_names
DiscourseBus.publish "/server-name", {channel: "/server-name-reply/#{my_id}"}
MessageBus.publish "/server-name", {channel: "/server-name-reply/#{my_id}"}
end
def self.seen_hosts
@ -20,12 +20,12 @@ class MessageBusDiags
unless @subscribed
DiscourseBus.subscribe "/server-name-reply/#{my_id}" do |msg|
MessageBus.subscribe "/server-name-reply/#{my_id}" do |msg|
MessageBusDiags.seen_host(msg.data)
end
DiscourseBus.subscribe "/server-name" do |msg|
DiscourseBus.publish msg.data["channel"], MessageBusDiags.my_id
MessageBus.subscribe "/server-name" do |msg|
MessageBus.publish msg.data["channel"], MessageBusDiags.my_id
end
@subscribed = true
end

View File

@ -221,7 +221,7 @@ module SiteSettingExtension
def ensure_listen_for_changes
unless @subscribed
DiscourseBus.subscribe("/site_settings") do |message|
MessageBus.subscribe("/site_settings") do |message|
process_message(message)
end
@subscribed = true
@ -233,10 +233,10 @@ module SiteSettingExtension
if data["process"] != process_id
begin
@last_message_processed = message.global_id
DiscourseBus.on_connect.call(message.site_id)
MessageBus.on_connect.call(message.site_id)
refresh!
ensure
DiscourseBus.on_disconnect.call(message.site_id)
MessageBus.on_disconnect.call(message.site_id)
end
end
end
@ -294,7 +294,7 @@ module SiteSettingExtension
end
def notify_changed!
DiscourseBus.publish('/site_settings', {process: process_id})
MessageBus.publish('/site_settings', {process: process_id})
end
def has_setting?(name)

View File

@ -70,7 +70,7 @@ after_initialize do
post.custom_fields["#{VOTES_CUSTOM_FIELD}-#{user_id}"] = votes
post.save_custom_fields(true)
DiscourseBus.publish("/polls/#{post_id}", { polls: polls })
MessageBus.publish("/polls/#{post_id}", { polls: polls })
return [poll, options]
end
@ -100,7 +100,7 @@ after_initialize do
post.save_custom_fields(true)
DiscourseBus.publish("/polls/#{post_id}", { polls: polls })
MessageBus.publish("/polls/#{post_id}", { polls: polls })
polls[poll_name]
end
@ -299,7 +299,7 @@ after_initialize do
post.save_custom_fields(true)
# publish the changes
DiscourseBus.publish("/polls/#{post.id}", { polls: polls })
MessageBus.publish("/polls/#{post.id}", { polls: polls })
end
end
else
@ -321,7 +321,7 @@ after_initialize do
# tells the front-end we have a poll for that post
on(:post_created) do |post|
next if post.is_first_post? || post.custom_fields[POLLS_CUSTOM_FIELD].blank?
DiscourseBus.publish("/polls", { post_id: post.id })
MessageBus.publish("/polls", { post_id: post.id })
end
add_to_serializer(:post, :polls, false) { post_custom_fields[POLLS_CUSTOM_FIELD] }

View File

@ -10,7 +10,7 @@ describe ::DiscoursePoll::PollsController do
describe "#vote" do
it "works" do
DiscourseBus.expects(:publish)
MessageBus.expects(:publish)
xhr :put, :vote, { post_id: poll.id, poll_name: "poll", options: ["5c24fc1df56d764b550ceae1b9319125"] }
@ -84,7 +84,7 @@ describe ::DiscoursePoll::PollsController do
describe "#toggle_status" do
it "works for OP" do
DiscourseBus.expects(:publish)
MessageBus.expects(:publish)
xhr :put, :toggle_status, { post_id: poll.id, poll_name: "poll", status: "closed" }
expect(response).to be_success
@ -94,7 +94,7 @@ describe ::DiscoursePoll::PollsController do
it "works for staff" do
log_in(:moderator)
DiscourseBus.expects(:publish)
MessageBus.expects(:publish)
xhr :put, :toggle_status, { post_id: poll.id, poll_name: "poll", status: "closed" }
expect(response).to be_success

View File

@ -89,7 +89,7 @@ describe Discourse do
it "adds a key in redis and publish a message through the message bus" do
$redis.expects(:set).with(Discourse.readonly_mode_key, 1)
DiscourseBus.expects(:publish).with(Discourse.readonly_channel, true)
MessageBus.expects(:publish).with(Discourse.readonly_channel, true)
Discourse.enable_readonly_mode
end
@ -99,7 +99,7 @@ describe Discourse do
it "removes a key from redis and publish a message through the message bus" do
$redis.expects(:del).with(Discourse.readonly_mode_key)
DiscourseBus.expects(:publish).with(Discourse.readonly_channel, false)
MessageBus.expects(:publish).with(Discourse.readonly_channel, false)
Discourse.disable_readonly_mode
end

View File

@ -72,7 +72,7 @@ describe PostCreator do
it "does not notify on system messages" do
admin = Fabricate(:admin)
messages = DiscourseBus.track_publish do
messages = MessageBus.track_publish do
p = PostCreator.create(admin, basic_topic_params.merge(post_type: Post.types[:moderator_action]))
PostCreator.create(admin, basic_topic_params.merge(topic_id: p.topic_id, post_type: Post.types[:moderator_action]))
end
@ -93,7 +93,7 @@ describe PostCreator do
created_post = nil
reply = nil
messages = DiscourseBus.track_publish do
messages = MessageBus.track_publish do
created_post = PostCreator.new(admin, basic_topic_params.merge(category: cat.id)).create
reply = PostCreator.new(admin, raw: "this is my test reply 123 testing", topic_id: created_post.topic_id).create
end
@ -118,7 +118,7 @@ describe PostCreator do
it 'generates the correct messages for a normal topic' do
p = nil
messages = DiscourseBus.track_publish do
messages = MessageBus.track_publish do
p = creator.create
end

View File

@ -573,7 +573,7 @@ describe Topic do
describe "make_banner!" do
it "changes the topic archetype to 'banner'" do
messages = DiscourseBus.track_publish do
messages = MessageBus.track_publish do
topic.make_banner!(user)
expect(topic.archetype).to eq(Archetype.banner)
end
@ -597,7 +597,7 @@ describe Topic do
it "resets the topic archetype" do
topic.expects(:add_moderator_post)
DiscourseBus.expects(:publish).with("/site/banner", nil)
MessageBus.expects(:publish).with("/site/banner", nil)
topic.remove_banner!(user)
expect(topic.archetype).to eq(Archetype.default)
end