Class: GraphQL::Subscriptions
- Inherits:
-
Object
- Object
- GraphQL::Subscriptions
- Defined in:
- lib/graphql/subscriptions.rb,
lib/graphql/subscriptions/event.rb,
lib/graphql/subscriptions/serialize.rb,
lib/graphql/subscriptions/instrumentation.rb,
lib/graphql/subscriptions/subscription_root.rb,
lib/graphql/subscriptions/broadcast_analyzer.rb,
lib/graphql/subscriptions/action_cable_subscriptions.rb,
lib/graphql/subscriptions/default_subscription_resolve_extension.rb
Direct Known Subclasses
Defined Under Namespace
Modules: Serialize, SubscriptionRoot Classes: ActionCableSubscriptions, BroadcastAnalyzer, DefaultSubscriptionResolveExtension, Event, Instrumentation, InvalidTriggerError
Instance Attribute Summary collapse
-
#default_broadcastable ⇒ Boolean
readonly
Used when fields don’t have
broadcastable:
explicitly set.
Class Method Summary collapse
Instance Method Summary collapse
-
#broadcastable?(query_str, **query_options) ⇒ Boolean
If true, then a query like this one would be broadcasted.
-
#build_id ⇒ String
A new unique identifier for a subscription.
-
#delete_subscription(subscription_id) ⇒ Object
A subscription was terminated server-side.
-
#deliver(subscription_id, result) ⇒ void
A subscription query was re-evaluated, returning
result
. -
#each_subscription_id(event) {|subscription_id| ... } ⇒ void
Get each
subscription_id
subscribed toevent.topic
and yield them. -
#execute(subscription_id, event, object) ⇒ void
Run the update query for this subscription and deliver it.
-
#execute_all(event, object) ⇒ void
Event
event
occurred onobject
, Update all subscribers. -
#execute_update(subscription_id, event, object) ⇒ GraphQL::Query::Result
event
was triggered onobject
, andsubscription_id
was subscribed, so it should be updated. -
#initialize(schema:, broadcast: false, default_broadcastable: false, **rest) ⇒ Subscriptions
constructor
A new instance of Subscriptions.
-
#normalize_name(event_or_arg_name) ⇒ String
Convert a user-provided event name or argument to the equivalent in GraphQL.
-
#read_subscription(subscription_id) ⇒ Hash
The system wants to send an update to this subscription.
-
#trigger(event_name, args, object, scope: nil) ⇒ void
Fetch subscriptions matching this field + arguments pair And pass them off to the queue.
-
#write_subscription(query, events) ⇒ void
query
was executed and found subscriptions toevents
.
Constructor Details
#initialize(schema:, broadcast: false, default_broadcastable: false, **rest) ⇒ Subscriptions
Returns a new instance of Subscriptions.
37 38 39 40 41 42 43 44 45 46 |
# File 'lib/graphql/subscriptions.rb', line 37 def initialize(schema:, broadcast: false, default_broadcastable: false, **rest) if broadcast if !schema.using_ast_analysis? raise ArgumentError, "`broadcast: true` requires AST analysis, add `using GraphQL::Analysis::AST` to your schema or see https://graphql-ruby.org/queries/ast_analysis.html." end schema.query_analyzer(Subscriptions::BroadcastAnalyzer) end @default_broadcastable = default_broadcastable @schema = schema end |
Instance Attribute Details
#default_broadcastable ⇒ Boolean (readonly)
Returns Used when fields don’t have broadcastable:
explicitly set.
49 50 51 |
# File 'lib/graphql/subscriptions.rb', line 49 def default_broadcastable @default_broadcastable end |
Class Method Details
.use(defn, options = {}) ⇒ Object
20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 |
# File 'lib/graphql/subscriptions.rb', line 20 def self.use(defn, = {}) schema = defn.is_a?(Class) ? defn : defn.target if schema.subscriptions raise ArgumentError, "Can't reinstall subscriptions. #{schema} is using #{schema.subscriptions}, can't also add #{self}" end instrumentation = Subscriptions::Instrumentation.new(schema: schema) defn.instrument(:query, instrumentation) defn.instrument(:field, instrumentation) [:schema] = schema schema.subscriptions = self.new(**) schema.add_subscription_extension_if_necessary nil end |
Instance Method Details
#broadcastable?(query_str, **query_options) ⇒ Boolean
Returns if true, then a query like this one would be broadcasted.
222 223 224 225 226 227 228 229 |
# File 'lib/graphql/subscriptions.rb', line 222 def broadcastable?(query_str, **) query = GraphQL::Query.new(@schema, query_str, **) if !query.valid? raise "Invalid query: #{query.validation_errors.map(&:to_h).inspect}" end GraphQL::Analysis::AST.analyze_query(query, @schema.query_analyzers) query.context.namespace(:subscriptions)[:subscription_broadcastable] end |
#build_id ⇒ String
Returns A new unique identifier for a subscription.
205 206 207 |
# File 'lib/graphql/subscriptions.rb', line 205 def build_id SecureRandom.uuid end |
#delete_subscription(subscription_id) ⇒ Object
A subscription was terminated server-side. Clean up the database.
200 201 202 |
# File 'lib/graphql/subscriptions.rb', line 200 def delete_subscription(subscription_id) raise GraphQL::RequiredImplementationMissingError end |
#deliver(subscription_id, result) ⇒ void
This method returns an undefined value.
A subscription query was re-evaluated, returning result
.
The result should be send to subscription_id
.
183 184 185 |
# File 'lib/graphql/subscriptions.rb', line 183 def deliver(subscription_id, result) raise GraphQL::RequiredImplementationMissingError end |
#each_subscription_id(event) {|subscription_id| ... } ⇒ void
This method returns an undefined value.
Get each subscription_id
subscribed to event.topic
and yield them
166 167 168 |
# File 'lib/graphql/subscriptions.rb', line 166 def each_subscription_id(event) raise GraphQL::RequiredImplementationMissingError end |
#execute(subscription_id, event, object) ⇒ void
This method returns an undefined value.
Run the update query for this subscription and deliver it
144 145 146 147 148 149 |
# File 'lib/graphql/subscriptions.rb', line 144 def execute(subscription_id, event, object) res = execute_update(subscription_id, event, object) if !res.nil? deliver(subscription_id, res) end end |
#execute_all(event, object) ⇒ void
This method returns an undefined value.
Event event
occurred on object
,
Update all subscribers.
156 157 158 159 160 |
# File 'lib/graphql/subscriptions.rb', line 156 def execute_all(event, object) each_subscription_id(event) do |subscription_id| execute(subscription_id, event, object) end end |
#execute_update(subscription_id, event, object) ⇒ GraphQL::Query::Result
event
was triggered on object
, and subscription_id
was subscribed,
so it should be updated.
Load subscription_id
’s GraphQL data, re-evaluate the query and return the result.
97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 |
# File 'lib/graphql/subscriptions.rb', line 97 def execute_update(subscription_id, event, object) # Lookup the saved data for this subscription query_data = read_subscription(subscription_id) if query_data.nil? delete_subscription(subscription_id) return nil end # Fetch the required keys from the saved data query_string = query_data.fetch(:query_string) variables = query_data.fetch(:variables) context = query_data.fetch(:context) operation_name = query_data.fetch(:operation_name) result = nil # this will be set to `false` unless `.execute` is terminated # with a `throw :graphql_subscription_unsubscribed` unsubscribed = true catch(:graphql_subscription_unsubscribed) do catch(:graphql_no_subscription_update) do # Re-evaluate the saved query, # but if it terminates early with a `throw`, # it will stay `nil` result = @schema.execute( query: query_string, context: context, subscription_topic: event.topic, operation_name: operation_name, variables: variables, root_value: object, ) end unsubscribed = false end if unsubscribed # `unsubscribe` was called, clean up on our side # TODO also send `{more: false}` to client? delete_subscription(subscription_id) end result end |
#normalize_name(event_or_arg_name) ⇒ String
Convert a user-provided event name or argument to the equivalent in GraphQL.
By default, it converts the identifier to camelcase. Override this in a subclass to change the transformation.
217 218 219 |
# File 'lib/graphql/subscriptions.rb', line 217 def normalize_name(event_or_arg_name) Schema::Member::BuildType.camelize(event_or_arg_name.to_s) end |
#read_subscription(subscription_id) ⇒ Hash
The system wants to send an update to this subscription. Read its data and return it.
174 175 176 |
# File 'lib/graphql/subscriptions.rb', line 174 def read_subscription(subscription_id) raise GraphQL::RequiredImplementationMissingError end |
#trigger(event_name, args, object, scope: nil) ⇒ void
This method returns an undefined value.
Fetch subscriptions matching this field + arguments pair And pass them off to the queue.
58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 |
# File 'lib/graphql/subscriptions.rb', line 58 def trigger(event_name, args, object, scope: nil) event_name = event_name.to_s # Try with the verbatim input first: field = @schema.get_field(@schema.subscription, event_name) if field.nil? # And if it wasn't found, normalize it: normalized_event_name = normalize_name(event_name) field = @schema.get_field(@schema.subscription, normalized_event_name) if field.nil? raise InvalidTriggerError, "No subscription matching trigger: #{event_name} (looked for #{@schema.subscription.graphql_name}.#{normalized_event_name})" end else # Since we found a field, the original input was already normalized normalized_event_name = event_name end # Normalize symbol-keyed args to strings, try camelizing them normalized_args = normalize_arguments(normalized_event_name, field, args) event = Subscriptions::Event.new( name: normalized_event_name, arguments: normalized_args, field: field, scope: scope, ) execute_all(event, object) end |
#write_subscription(query, events) ⇒ void
This method returns an undefined value.
query
was executed and found subscriptions to events
.
Update the database to reflect this new state.
192 193 194 |
# File 'lib/graphql/subscriptions.rb', line 192 def write_subscription(query, events) raise GraphQL::RequiredImplementationMissingError end |