Packages
absinthe
1.5.0-alpha.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 false
require Logger
alias Absinthe.Pipeline.BatchResolver
# This module handles running and broadcasting documents that are local to this
# node.
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, {field, key_strategy}}, put_in(doc.execution.root_value, mutation_result)}
end
docs_by_context = group_by_context(docs_and_topics)
for docset <- docs_by_context do
run_docset(pubsub, docset)
end
end
defp group_by_context(docs_and_topics) do
docs_and_topics
|> Enum.group_by(fn {_, doc} -> doc.execution.context end)
|> Map.values()
end
defp run_docset(pubsub, docs_and_topics) do
{topics, docs} = Enum.unzip(docs_and_topics)
docs = BatchResolver.run(docs, schema: hd(docs).schema, abort_on_error: false)
pipeline = [
Absinthe.Phase.Document.Result
]
for {doc, {topic, key_strategy}} <- Enum.zip(docs, topics), doc != :error do
try do
{:ok, %{result: data}, _} = Absinthe.Pipeline.run(doc, 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, System.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