deliver messages asynchronously

Author: prog-supdex

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"

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"
 

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

lib/graphql/subscriptions/adapters/base_job.rb

@@ -0,0 +1,17 @@
+# frozen_string_literal: true
+
+require "active_job"
+
+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

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