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.
39 40 41 42 43 44 45 46 47 48 |
# File 'lib/graphql/subscriptions.rb', line 39 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.
51 52 53 |
# File 'lib/graphql/subscriptions.rb', line 51 def default_broadcastable @default_broadcastable end |
Class Method Details
.use(defn, options = {}) ⇒ Object
22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 |
# File 'lib/graphql/subscriptions.rb', line 22 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.
224 225 226 227 228 229 230 231 |
# File 'lib/graphql/subscriptions.rb', line 224 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.
207 208 209 |
# File 'lib/graphql/subscriptions.rb', line 207 def build_id SecureRandom.uuid end |
#delete_subscription(subscription_id) ⇒ Object
A subscription was terminated server-side. Clean up the database.
202 203 204 |
# File 'lib/graphql/subscriptions.rb', line 202 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
.
185 186 187 |
# File 'lib/graphql/subscriptions.rb', line 185 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
168 169 170 |
# File 'lib/graphql/subscriptions.rb', line 168 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
146 147 148 149 150 151 |
# File 'lib/graphql/subscriptions.rb', line 146 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.
158 159 160 161 162 |
# File 'lib/graphql/subscriptions.rb', line 158 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.
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 139 140 |
# File 'lib/graphql/subscriptions.rb', line 99 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.
219 220 221 |
# File 'lib/graphql/subscriptions.rb', line 219 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.
176 177 178 |
# File 'lib/graphql/subscriptions.rb', line 176 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.
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 87 88 |
# File 'lib/graphql/subscriptions.rb', line 60 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.
194 195 196 |
# File 'lib/graphql/subscriptions.rb', line 194 def write_subscription(query, events) raise GraphQL::RequiredImplementationMissingError end |