diff --git a/graphql-anycable.gemspec b/graphql-anycable.gemspec index a5496e0..4041b3e 100644 --- a/graphql-anycable.gemspec +++ b/graphql-anycable.gemspec @@ -30,6 +30,7 @@ Gem::Specification.new do |spec| spec.add_dependency "anyway_config", ">= 1.3", "< 3" spec.add_dependency "graphql", ">= 1.11", "< 3" spec.add_dependency "redis", ">= 4.2.0" + spec.add_dependency "activejob", ">= 5.0.0" spec.add_development_dependency "anycable-rails" spec.add_development_dependency "bundler", "~> 2.0" diff --git a/lib/graphql-anycable.rb b/lib/graphql-anycable.rb index 97d18f1..92fa66a 100644 --- a/lib/graphql-anycable.rb +++ b/lib/graphql-anycable.rb @@ -6,6 +6,7 @@ require_relative "graphql/anycable/cleaner" require_relative "graphql/anycable/config" require_relative "graphql/anycable/railtie" if defined?(Rails) +require_relative "graphql/subscriptions/adapters/base_job" require_relative "graphql/anycable/stats" require_relative "graphql/subscriptions/anycable_subscriptions" diff --git a/lib/graphql/anycable/config.rb b/lib/graphql/anycable/config.rb index c422ca6..2816c4c 100644 --- a/lib/graphql/anycable/config.rb +++ b/lib/graphql/anycable/config.rb @@ -12,6 +12,9 @@ class Config < Anyway::Config attr_config use_redis_object_on_cleanup: true attr_config use_client_provided_uniq_id: true attr_config redis_prefix: "graphql" # Here, we set clear redis_prefix without any hyphen. The hyphen is added at the end of this value on our side. + + attr_config use_async_broadcasting: true + attr_config async_broadcasting: { queue: "broadcasting", class: "GraphQL::Adapters::BaseJob" } end end end diff --git a/lib/graphql/subscriptions/adapters/base_job.rb b/lib/graphql/subscriptions/adapters/base_job.rb new file mode 100644 index 0000000..acea3ce --- /dev/null +++ b/lib/graphql/subscriptions/adapters/base_job.rb @@ -0,0 +1,15 @@ +# frozen_string_literal: true + +module GraphQL + module Adapters + class BaseJob < ActiveJob::Base + DEFAULT_QUEUE_NAME = :default + + queue_as { GraphQL::AnyCable.config.async_broadcasting["queue"] || DEFAULT_QUEUE_NAME } + + def perform(subscription_object, execute_method, event, object) + Marshal.load(subscription_object).public_send(execute_method, Marshal.load(event), Marshal.load(object)) + end + end + end +end diff --git a/lib/graphql/subscriptions/anycable_subscriptions.rb b/lib/graphql/subscriptions/anycable_subscriptions.rb index 6dec5d8..1dfc291 100644 --- a/lib/graphql/subscriptions/anycable_subscriptions.rb +++ b/lib/graphql/subscriptions/anycable_subscriptions.rb @@ -60,6 +60,7 @@ class AnyCableSubscriptions < GraphQL::Subscriptions FINGERPRINTS_PREFIX = "fingerprints:" # ZSET: To get fingerprints by topic SUBSCRIPTIONS_PREFIX = "subscriptions:" # SET: To get subscriptions by fingerprint CHANNEL_PREFIX = "channel:" # SET: Auxiliary structure for whole channel's subscriptions cleanup + EXECUTOR_METHOD_NAME = "execute_synchronically" # method, who execute the main logic # @param serializer [<#dump(obj), #load(string)] Used for serializing messages before handing them to `.broadcast(msg)` def initialize(serializer: Serialize, **rest) @@ -73,6 +74,13 @@ def execute_all(event, object) fingerprints = redis.zrange(redis_key(FINGERPRINTS_PREFIX) + event.topic, 0, -1) return if fingerprints.empty? + perform(event, object) + end + + def execute_synchronically(event, object) + fingerprints = redis.zrange(redis_key(FINGERPRINTS_PREFIX) + event.topic, 0, -1) + return if fingerprints.empty? + fingerprint_subscription_ids = Hash[fingerprints.zip( redis.pipelined do |pipeline| fingerprints.map do |fingerprint| @@ -243,6 +251,24 @@ def fetch_channel_istate(channel) def redis_key(prefix) "#{config.redis_prefix}-#{prefix}" end + + def executor_class_job + custom_class = config.async_broadcasting["class"] + + return Adapters::BaseJob unless custom_class + + Object.const_get(config.async_broadcasting["class"]) + end + + def perform(event, object) + unless config.use_async_broadcasting + return public_send(EXECUTOR_METHOD_NAME, event, object) + end + + args = [Marshal.dump(self), EXECUTOR_METHOD_NAME, Marshal.dump(event), Marshal.dump(object)] + + executor_class_job.perform_later(*args) + end end end end