Packages
absinthe
1.7.7
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_fn =
if function_exported?(pubsub, :run_docset, 3), do: &pubsub.run_docset/3, else: &run_docset/3
run_docset_fn.(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 = pipeline(doc, mutation_result)
{: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
def pipeline(doc, mutation_result) 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,
[
result_phase(doc),
{Absinthe.Phase.Telemetry, event: [:subscription, :publish, :stop]}
]
]
pipeline
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
defp result_phase(doc) do
# use the configured result phase from the initial pipeline
# this will allow the result of the subscription data to match
# the output of query/mutation. An example of result phase is
# Absinthe.Phoenix.Controller.Result where the output will have
# atom keys and allow struct to be returned
doc.initial_phases
|> Pipeline.from(Phase.Blueprint)
|> case do
[{Phase.Blueprint, opts} | _] ->
Keyword.get(opts, :result_phase, Phase.Document.Result)
_ ->
Phase.Document.Result
end
end
end