retort

v2.7.0

retort v2.7.0 Retort.Client.Generic View Source

A generic client for interacting with a resource over JSON API over JSON RPC over RabbitMQ.

Link to this section Summary

Types

error()

Common tagged tuple return format for errors

id()

id field of a resource

queue()

The name of a queue

resource()

A model struct for a resource

t()
  • channel - The channel connected to the RabbitMQ broker. Created from the connection passed to start_link/2.
  • consumer_tag - The consumer tag assigned to the process when it consumes its reply_queue on channel
  • ecto_schema_module_by_type - Maps JSON API type field to the corresponding Ecto.Schema Module for building resources from Retort.Response results.
  • meta - Meta information about jsonapi request.
  • reply_queue - The name of queue on which this server gets replies
  • request_by_correlation_id - Tracks the from (calling pid and reference) that called a client function and which method was called as teh method influences the allowed formats for responses.
  • request_queue - The name of the queue on which this server publishes requests. Matches queue passed to start_link/2.
  • type - Type used for create/2
  • type_by_ecto_schema_module - Maps model back to JSON API type for relationship "type" field. Automatically derived from ecto_schema_module_by_type
type_by_ecto_schema_module()

Maps Ecto.Schema Module for building resources back to JSON API type field, so the inverse of IntepreterServer.Api.ToEctoSchema.ecto_schema_module_by_type

Functions

count(pid, params \\ %{}, timeout \\ 5000)

Counts the number of resources

create(pid, convertable_params, mergable_params \\ %{}, timeout \\ 5000)

Creates a resource

destroy(pid, id, timeout \\ 5000)

Destroys a resource by id

destroy_all(pid, params \\ %{}, timeout \\ 5000)

Destroys all resources. Stops on first error

error_to_calcinator_error(error)

Converts Retort.Client.Generic.error to errors returned by Calcinator and Calcinator.Resources callbacks

error_to_calcinator_error(arg, changeset)

Converts Retort.Client.Generic.error to errors returned by Calcinator and Calcinator.Resources callbacks

handle_info(msg, state)

The consumer process will receive the following data structures

index(pid, params \\ %{}, timeout \\ 5000)

Retrieves the index of this client’s resource

init(args)

Creates a channel and anonymous queue for this GenServer to receive replies on

show(pid, id, params \\ %{}, timeout \\ 5000)

Retrieves resource with the given id

start_link(opts \\ [])

Starts a client pid to send requests and wait for responses. To have more than one request in flight, you need multiple clients from multiple calls to this function

stop(pid)

Stops a client’s GenServer that is waiting for responses

stream(pid, params \\ %{}, timeout \\ 5000)

Streams the index of this client’s resource, automatically fetching the next page

terminate(reason, arg2)

Closes the channel this GenServer instance started

update(pid, id, convertable_params, mergable_params \\ %{}, timeout \\ 5000)

Update the attributes and/or associations of resource with the given id

Link to this section Types

Link to this type error() View Source 
error() :: {:error, Retort.Response.Error.t()}

Common tagged tuple return format for errors.

Link to this type id() View Source 
id() :: String.t()

id field of a resource

Link to this type queue() View Source 
queue() :: String.t()

The name of a queue

Link to this type resource() View Source 
resource() :: struct()

A model struct for a resource

Link to this type t() View Source 
t() :: %Retort.Client.Generic{channel: %AMQP.Channel{conn: term(), pid: term()}, consumer_tag: String.t(), ecto_schema_module_by_type: Alembic.ToEctoSchema.ecto_schema_module_by_type(), meta: map(), reply_queue: String.t(), request_by_correlation_id: %{optional(String.t()) => ClientRequest.t()}, request_queue: String.t(), type: String.t(), type_by_ecto_schema_module: type_by_ecto_schema_module()}
  • channel - The channel connected to the RabbitMQ broker. Created from the connection passed to start_link/2.
  • consumer_tag - The consumer tag assigned to the process when it consumes its reply_queue on channel
  • ecto_schema_module_by_type - Maps JSON API type field to the corresponding Ecto.Schema Module for building resources from Retort.Response results.
  • meta - Meta information about jsonapi request.
  • reply_queue - The name of queue on which this server gets replies
  • request_by_correlation_id - Tracks the from (calling pid and reference) that called a client function and which method was called as teh method influences the allowed formats for responses.
  • request_queue - The name of the queue on which this server publishes requests. Matches queue passed to start_link/2.
  • type - Type used for create/2
  • type_by_ecto_schema_module - Maps model back to JSON API type for relationship "type" field. Automatically derived from ecto_schema_module_by_type.
Link to this type type_by_ecto_schema_module() View Source 
type_by_ecto_schema_module() :: %{optional(Alembic.ToEctoSchema.ecto_schema_module()) => Alembic.Resource.type()}

Maps Ecto.Schema Module for building resources back to JSON API type field, so the inverse of IntepreterServer.Api.ToEctoSchema.ecto_schema_module_by_type.

Link to this section Functions

Link to this function count(pid, params \\ %{}, timeout \\ 5000) View Source 
count(pid(), map(), timeout()) :: non_neg_integer()

Counts the number of resources

Link to this function create(pid, convertable_params, mergable_params \\ %{}, timeout \\ 5000) View Source 
create(pid(), map(), map(), timeout()) :: {:ok, resource()} | error()

Creates a resource.

  • pid - the pid of the client
  • convertable_params - the nested params to be converted to the data and relationships in the JSON API document. All keys should be String.t and NOT atom.
  • mergable_params - additional params to be merged with the JSON API document created from convertable_params. This allows setting of include query params or meta in the document.

All keys in convertable_params and mergable_params should be String.t and NOT atom.

Link to this function destroy(pid, id, timeout \\ 5000) View Source 
destroy(pid(), id(), timeout()) :: :ok | error()

Destroys a resource by id

Link to this function destroy_all(pid, params \\ %{}, timeout \\ 5000) View Source 
destroy_all(pid(), map(), timeout()) :: :ok | error()

Destroys all resources. Stops on first error.

Link to this function error_to_calcinator_error(error) View Source 
error_to_calcinator_error(error()) ::
  {:error, :bad_gateway} |
  {:error, :not_found} |
  {:error, :sandbox_access_disallowed} |
  {:error, Ecto.Changeset.t()}

Converts Retort.Client.Generic.error to errors returned by Calcinator and Calcinator.Resources callbacks.

Link to this function error_to_calcinator_error(arg, changeset) View Source 
error_to_calcinator_error(error(), maybe_changeset :: Ecto.Changeset.t() | nil) ::
  {:error, :bad_gateway} |
  {:error, :not_found} |
  {:error, :sandbox_access_disallowed} |
  {:error, Ecto.Changeset.t()}

Converts Retort.Client.Generic.error to errors returned by Calcinator and Calcinator.Resources callbacks.

Link to this function handle_info(msg, state) View Source 
handle_info({:basic_deliver, String.t(), %{correlationed_id: String.t(), delivery_tag: binary()}}, t()) :: {:noreply, t()}
handle_info({:basic_consume_ok, %{:consumer_tag => String.t(), optional(any()) => any()}}, t()) :: {:noreply, t()}
handle_info({:basic_consume, %{:consumer_tag => String.t(), optional(any()) => any()}}, t()) :: {:noreply, t()}
handle_info({:basic_cancel_ok, %{:consumer_tag => String.t(), optional(any()) => any()}}, t()) :: {:noreply, t()}
handle_info({:basic_cancel, %{:consumer_tag => String.t(), optional(any()) => any()}}, t()) :: {:stop, :normal, t()}

The consumer process will receive the following data structures:

  • {:basic_deliver, payload, meta} - This is sent for each message consumed, where payload contains the message content and meta contains all the metadata set when sending with AMQP.Basic.publish/5 or additional info set by the broker.
  • {:basic_consume_ok, %{consumer_tag: consumer_tag}} - Sent when the consumer process is registered with AMQP.Basic.consume/4. The caller receives the same information as the return of AMQP.Basic.consume/4.
  • {:basic_cancel, %{consumer_tag: consumer_tag, no_wait: no_wait}} - Sent by the broker when the consumer is unexpectedly cancelled (such as after a queue deletion)
  • {:basic_cancel_ok, %{consumer_tag: consumer_tag}} - Sent to the consumer process after a call to AMQP.Basic.cancel
Link to this function index(pid, params \\ %{}, timeout \\ 5000) View Source 
index(pid(), map(), timeout()) ::
  {:ok, [resource()], Alembic.Pagination.t() | nil} |
  error()

Retrieves the index of this client’s resource.

Link to this function init(args) View Source 
init([]) :: {:ok, t()}

Creates a channel and anonymous queue for this GenServer to receive replies on.

  • :connection (Retort.Connection.await/0) - %AMQP.Connection{} connection.
  • :ecto_schema_module_by_type - Maps JSON API type field to the corresponsing Ecto.Schema Module for building resources from Retort.Response results.
  • :meta Free form meta data about the JSONAPI request.
  • :queue - The name of the queue on which this server publishes requests. Matches queue passed to start_link/2.
  • :type - The primary JSON API "type", such as for create/2.
Link to this function show(pid, id, params \\ %{}, timeout \\ 5000) View Source 
show(pid(), id(), map(), timeout()) :: {:ok, resource()} | error()

Retrieves resource with the given id.

Link to this function start_link(opts \\ []) View Source 
start_link([connection: %AMQP.Connection{pid: term()}, ecto_schema_module_by_type: Alembic.ToEctoSchema.ecto_schema_module_by_type(), meta: map(), queue: queue(), ecto_repo_module: module(), type: String.t()]) :: GenServer.on_start()

Starts a client pid to send requests and wait for responses. To have more than one request in flight, you need multiple clients from multiple calls to this function.

  • :connection (Retort.Connection.await/0) - %AMQP.Connection{} connection.
  • :ecto_schema_module_by_type - Maps JSON API type field to the corresponsing Ecto.Schema Module for building resources from Retort.Response results.
  • :meta - Free form data structure to pass meta information in rpc request. Defaults to setting %{host: } from application env var. MUST pass Retort.Meta.valid!/2 if there are databased-backed ecto_schema_module in :ecto_schema_module_by_type and :ecto_repo_module is sandboxed.
  • :queue - The name of the queue on which this server publishes requests. Matches queue passed to start_link/2.
  • :ecto_repo_module - The Ecto.Repo whose Ecto.Repo.config/0 :pool to check if it is Ecto.Adapters.SQL.Sandbox if any of ecto_schema_module in ecto_schema_module_by_type has a __schema__(:source) set, which is assumed to mean they are backed by this Ect.Repo.t and so would be sandboxed.
  • :type - The primary JSON API "type", such as for create/2.
Link to this function stop(pid) View Source 
stop(pid()) :: :ok

Stops a client’s GenServer that is waiting for responses.

Starting and stopping a Retort.Client.Generic

iex> {:ok, pid} = Retort.Client.Generic.start_link("generic")
iex> Retort.Client.Partner.stop(pid)
:ok
Link to this function stream(pid, params \\ %{}, timeout \\ 5000) View Source 
stream(pid(), map(), timeout()) :: Enumerable.t()

Streams the index of this client’s resource, automatically fetching the next page.

Because the generated stream may make multiple requests, the timeout is the timeout for each individual index/3 request.

NOTE: DO NOT destroy/3 the returned resources if you plan to continuing using the stream as it will cause the cached pagination information to be invalid as you may have deleted enough resources that the page no longer exists.

If you need to destroy resources from a stream do one of the following:

  1. Convert it to a list, so all the remote index/3 requests are complete
  2. Use destroy_all/3 to delete the first page repeatedly until all resources are deleted.

Convert Stream to List Before Destroying

(1) has the benefit that only the resources you can see at the time of the call to Enum.to_list would be destoryed, but you need to download all the resources at once into a list.

iex> pid
...> |> Retort.Client.Generic.stream
...> |> Enum.to_list
...> |> Enum.each(fn %{id: id} ->
...>      Retort.Client.Generic.destroy(to_string(id))
...>    end)

Use destroy_all/3 

iex> Retort.Client.Generic.destroy_all(pid)

(2) only loads a page of resources at a time, but can never finish if new resources are being created that keep the first page with at least one resource when it rechecks.

Link to this function terminate(reason, arg2) View Source 
terminate(any(), t()) :: :ok

Closes the channel this GenServer instance started

Link to this function update(pid, id, convertable_params, mergable_params \\ %{}, timeout \\ 5000) View Source 
update(pid(), id(), map(), map(), timeout()) ::
  {:ok, resource()} |
  error()

Update the attributes and/or associations of resource with the given id

  • pid - the pid of the client
  • id - the id of the resource to update as a String.t.
  • convertable_params - the nested params to be converted to the data and relationships in the JSON API document. All keys should be String.t and NOT atom.
  • mergable_params - additional params to be merged with the JSON API document created from convertable_params. This allows setting of include query params or meta in the document.

All keys in convertable_params and mergable_params should be String.t and NOT atom.