Packages
absinthe
1.6.2
1.11.0
1.10.2
1.10.1
1.10.0
1.9.1
1.9.0
1.8.0
1.7.11
1.7.10
1.7.9
1.7.8
1.7.7
1.7.6
1.7.5
1.7.4
1.7.3
1.7.2
1.7.1
1.7.0
1.6.8
1.6.7
retired
1.6.6
1.6.5
1.6.4
1.6.3
1.6.2
1.6.1
1.6.0
1.6.0-rc.1
1.6.0-rc.0
1.5.5
1.5.4
1.5.3
1.5.2
1.5.1
1.5.0
1.5.0-rc.5
1.5.0-rc.4
1.5.0-rc.3
1.5.0-rc.2
1.5.0-rc.1
1.5.0-rc.0
1.5.0-beta.2
1.5.0-beta.1
1.5.0-beta.0
1.5.0-alpha.4
1.5.0-alpha.3
1.5.0-alpha.2
1.5.0-alpha.1
1.5.0-alpha.0
1.4.16
1.4.15
1.4.14
1.4.13
1.4.12
1.4.11
1.4.10
1.4.9
1.4.8
retired
1.4.7
1.4.6
1.4.5
1.4.4
1.4.3
1.4.2
1.4.1
1.4.0
1.4.0-rc.3
1.4.0-rc.2
1.4.0-rc.1
1.4.0-rc.0
1.4.0-beta.5
1.4.0-beta.4
1.4.0-beta.3
1.4.0-beta.2
1.4.0-beta.1
1.3.2
1.3.1
1.3.0
1.3.0-rc.0
1.3.0-beta.2
1.3.0-beta.1
1.3.0-beta.0
1.2.6
1.2.5
1.2.4
1.2.3
1.2.2
1.2.1
1.2.0
1.2.0-rc.0
1.2.0-beta.0
1.2.0-alpha0
1.2.0-alpha.2
1.2.0-alpha.1
1.1.11
1.1.10
1.1.9
1.1.8
1.1.7
1.1.6
1.1.5
1.1.4
1.1.3
1.1.2
1.1.1
1.1.0
1.0.0
0.5.2
0.5.1
0.5.0
0.4.6
0.4.5
0.4.4
0.4.3
0.4.2
0.4.1
0.4.0
0.2.3
0.2.2
0.2.1
0.1.0
GraphQL for Elixir
Security advisory:
This version has known vulnerabilities.
View advisories
Current section
Files
Jump to
Current section
Files
lib/absinthe/subscription/local.ex
defmodule Absinthe.Subscription.Local do
@moduledoc """
This module handles broadcasting documents that are local to this node
"""
require Logger
alias Absinthe.Pipeline.BatchResolver
# This module handles running and broadcasting documents that are local to this
# node.
@doc """
Publish a mutation to the local node only.
See also `Absinthe.Subscription.publish/3`
"""
@spec publish_mutation(
Absinthe.Subscription.Pubsub.t(),
term,
[Absinthe.Subscription.subscription_field_spec()]
) :: :ok
def publish_mutation(pubsub, mutation_result, subscribed_fields) do
docs_and_topics =
for {field, key_strategy} <- subscribed_fields,
{topic, doc} <- get_docs(pubsub, field, mutation_result, key_strategy) do
{topic, key_strategy, doc}
end
run_docset(pubsub, docs_and_topics, mutation_result)
:ok
end
alias Absinthe.{Phase, Pipeline}
defp run_docset(pubsub, docs_and_topics, mutation_result) do
for {topic, key_strategy, doc} <- docs_and_topics do
try do
pipeline =
doc.initial_phases
|> Pipeline.replace(
Phase.Telemetry,
{Phase.Telemetry, event: [:subscription, :publish, :start]}
)
|> Pipeline.without(Phase.Subscription.SubscribeSelf)
|> Pipeline.insert_before(
Phase.Document.Execution.Resolution,
{Phase.Document.OverrideRoot, root_value: mutation_result}
)
|> Pipeline.upto(Phase.Document.Execution.Resolution)
pipeline = [
pipeline,
[
Absinthe.Phase.Document.Result,
{Absinthe.Phase.Telemetry, event: [:subscription, :publish, :stop]}
]
]
{:ok, %{result: data}, _} = Absinthe.Pipeline.run(doc.source, pipeline)
Logger.debug("""
Absinthe Subscription Publication
Field Topic: #{inspect(key_strategy)}
Subscription id: #{inspect(topic)}
Data: #{inspect(data)}
""")
:ok = pubsub.publish_subscription(topic, data)
rescue
e ->
BatchResolver.pipeline_error(e, __STACKTRACE__)
end
end
end
defp get_docs(pubsub, field, mutation_result, topic: topic_fun)
when is_function(topic_fun, 1) do
do_get_docs(pubsub, field, topic_fun.(mutation_result))
end
defp get_docs(pubsub, field, _mutation_result, key) do
do_get_docs(pubsub, field, key)
end
defp do_get_docs(pubsub, field, keys) do
keys
|> List.wrap()
|> Enum.map(&to_string/1)
|> Enum.flat_map(&Absinthe.Subscription.get(pubsub, {field, &1}))
end
end