1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
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
87
88
89
90
91
92
93
94
95
96
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
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
|
# frozen_string_literal: true
module GraphQL
class Schema
# This class can be extended to create fields on your subscription root.
#
# It provides hooks for the different parts of the subscription lifecycle:
#
# - `#authorized?`: called before initial subscription and subsequent updates
# - `#subscribe`: called for the initial subscription
# - `#update`: called for subsequent update
#
# Also, `#unsubscribe` terminates the subscription.
class Subscription < GraphQL::Schema::Resolver
extend GraphQL::Schema::Resolver::HasPayloadType
extend GraphQL::Schema::Member::HasFields
NO_UPDATE = :no_update
# The generated payload type is required; If there's no payload,
# propagate null.
null false
def initialize(object:, context:, field:)
super
# Figure out whether this is an update or an initial subscription
@mode = context.query.subscription_update? ? :update : :subscribe
end
def resolve_with_support(**args)
result = nil
unsubscribed = true
unsubscribed_result = catch :graphql_subscription_unsubscribed do
result = super
unsubscribed = false
end
if unsubscribed
if unsubscribed_result
context.namespace(:subscriptions)[:final_update] = true
unsubscribed_result
else
context.skip
end
else
result
end
end
# Implement the {Resolve} API
def resolve(**args)
# Dispatch based on `@mode`, which will raise a `NoMethodError` if we ever
# have an unexpected `@mode`
public_send("resolve_#{@mode}", **args)
end
# Wrap the user-defined `#subscribe` hook
def resolve_subscribe(**args)
ret_val = args.any? ? subscribe(**args) : subscribe
if ret_val == :no_response
context.skip
else
ret_val
end
end
# The default implementation returns nothing on subscribe.
# Override it to return an object or
# `:no_response` to (explicitly) return nothing.
def subscribe(args = {})
:no_response
end
# Wrap the user-provided `#update` hook
def resolve_update(**args)
ret_val = args.any? ? update(**args) : update
if ret_val == NO_UPDATE
context.namespace(:subscriptions)[:no_update] = true
context.skip
else
ret_val
end
end
# The default implementation returns the root object.
# Override it to return {NO_UPDATE} if you want to
# skip updates sometimes. Or override it to return a different object.
def update(args = {})
object
end
# If an argument is flagged with `loads:` and no object is found for it,
# remove this subscription (assuming that the object was deleted in the meantime,
# or that it became inaccessible).
def load_application_object_failed(err)
if @mode == :update
unsubscribe
end
super
end
# Call this to halt execution and remove this subscription from the system
# @param update_value [Object] if present, deliver this update before unsubscribing
# @return [void]
def unsubscribe(update_value = nil)
context.namespace(:subscriptions)[:unsubscribed] = true
throw :graphql_subscription_unsubscribed, update_value
end
READING_SCOPE = ::Object.new
# Call this method to provide a new subscription_scope; OR
# call it without an argument to get the subscription_scope
# @param new_scope [Symbol]
# @param optional [Boolean] If true, then don't require `scope:` to be provided to updates to this subscription.
# @return [Symbol]
def self.subscription_scope(new_scope = READING_SCOPE, optional: false)
if new_scope != READING_SCOPE
@subscription_scope = new_scope
@subscription_scope_optional = optional
elsif defined?(@subscription_scope)
@subscription_scope
else
find_inherited_value(:subscription_scope)
end
end
def self.subscription_scope_optional?
if defined?(@subscription_scope_optional)
@subscription_scope_optional
else
find_inherited_value(:subscription_scope_optional, false)
end
end
# This is called during initial subscription to get a "name" for this subscription.
# Later, when `.trigger` is called, this will be called again to build another "name".
# Any subscribers with matching topic will begin the update flow.
#
# The default implementation creates a string using the field name, subscription scope, and argument keys and values.
# In that implementation, only `.trigger` calls with _exact matches_ result in updates to subscribers.
#
# To implement a filtered stream-type subscription flow, override this method to return a string with field name and subscription scope.
# Then, implement {#update} to compare its arguments to the current `object` and return {NO_UPDATE} when an
# update should be filtered out.
#
# @see {#update} for how to skip updates when an event comes with a matching topic.
# @param arguments [Hash<String => Object>] The arguments for this topic, in GraphQL-style (camelized strings)
# @param field [GraphQL::Schema::Field]
# @param scope [Object, nil] A value corresponding to `.trigger(... scope:)` (for updates) or the `subscription_scope` found in `context` (for initial subscriptions).
# @return [String] An identifier corresponding to a stream of updates
def self.topic_for(arguments:, field:, scope:)
Subscriptions::Serialize.dump_recursive([scope, field.graphql_name, arguments])
end
end
end
end
|
