Freddy.RPC.Client behaviour (freddy v0.17.1)

This module allows to build RPC client for any Freddy-compliant microservice.

Example

defmodule PaymentsService do
  use Freddy.RPC.Client

  @config [timeout: 3500]

  def start_link(conn, initial, opts \\ []) do
    Freddy.RPC.Client.start_link(__MODULE__, conn, @config, initial, opts)
  end
end

{:ok, client} = PaymentsService.start_link()
PaymentsService.request(client, "Payments", %{type: "get_history", site_id: "xxx"})

Link to this section Summary

Types

config()

meta()

opts()

payload()

request()

response()

routing_key()

state()

Functions

call(client, message, timeout \\ 5000)

See Connection.call/3.

cast(client, message)

See Connection.cast/2.

request(client, routing_key, payload, options \\ [])

Performs a RPC request and blocks until the response arrives.

start(mod, connection, config, initial, options \\ [])

Starts a Freddy.RPC.Client process without linking to the current process, see start_link/5 for more information.

start_link(mod, connection, config, initial, options \\ [])

Starts a Freddy.RPC.Client process linked to the current process.

stop(client, reason \\ :normal)

See GenServer.stop/2.

Callbacks

before_request(request, state)

Called before a request will be performed to the exchange.

decode_response(payload, meta, request, state)

Called when a response message is delivered from the queue before passing it into a on_response function.

encode_request(request, state)

Called before a message will be published to the exchange.

handle_call(request, arg2, state)

Called when the process receives a call message sent by call/3. This callback has the same arguments as the GenServer equivalent and the :reply, :noreply and :stop return tuples behave the same.

handle_cast(request, state)

Called when the process receives a cast message sent by cast/2. This callback has the same arguments as the GenServer equivalent and the :noreply and :stop return tuples behave the same.

handle_connected(arg1, state)

Called when the RPC client process has opened AMQP channel before registering itself as a consumer.

handle_disconnected(reason, state)

Called when the AMQP server has been disconnected from the AMQP broker.

handle_info(message, state)

Called when the process receives a message. This callback has the same arguments as the GenServer equivalent and the :noreply and :stop return tuples behave the same.

handle_ready(meta, state)

Called when the AMQP server has registered the process as a consumer of the server-named queue and it will start to receive messages.

init(initial)

Called when the RPC client process is first started. start_link/5 will block until it returns.

on_response(response, request, state)

Called when a response has been received, before it is delivered to the caller.

on_return(request, state)

Called when a request has been returned by AMPQ broker.

on_timeout(request, state)

Called when a request has timed out.

terminate(reason, state)

This callback is the same as the GenServer equivalent and is called when the process terminates. The first argument is the reason the process is about to exit with.

Link to this section Types

config()

Specs

config() :: [timeout: timeout(), exchange: Keyword.t()]

meta()

Specs

meta() :: map()

opts()

Specs

opts() :: Keyword.t()

payload()

Specs

payload() :: term()

request()

Specs

request() :: Freddy.RPC.Request.t()

response()

Specs

response() :: term()

routing_key()

Specs

routing_key() :: String.t()

state()

Specs

state() :: term()

Link to this section Functions

call(client, message, timeout \\ 5000)

See Connection.call/3.

cast(client, message)

See Connection.cast/2.

request(client, routing_key, payload, options \\ [])

Specs

request(GenServer.server(), routing_key(), payload(), Keyword.t()) ::
  {:ok, response()}
  | {:error, reason :: term()}
  | {:error, reason :: term(), hint :: term()}

Performs a RPC request and blocks until the response arrives.

The routing_key parameter specifies the routing key for the message. The routing key is used by the RabbitMQ server to route a message from an exchange to worker queues or another exchanges.

The payload parameter specifies the message content as an Erlang term. The payload is converted to binary string by the encode_request/2 callback before sending it to server.

Options

:timeout - if present, the client is allowed to wait given number of milliseconds for the response message from the server
:headers - message headers
:persistent - if set, uses persistent delivery mode
:priority - message priority, ranging from 0 to 9
:message_id - message identifier
:timestamp - timestamp associated with this message (epoch time)
:user_id - creating user ID. RabbitMQ will validate this against the active connection user
:app_id - publishing application ID

start(mod, connection, config, initial, options \\ [])

Specs

start(
  module(),
  GenServer.server(),
  config(),
  initial :: term(),
  GenServer.options()
) :: GenServer.on_start()

Starts a Freddy.RPC.Client process without linking to the current process, see start_link/5 for more information.

start_link(mod, connection, config, initial, options \\ [])

Specs

start_link(
  module(),
  GenServer.server(),
  config(),
  initial :: term(),
  GenServer.options()
) :: GenServer.on_start()

Starts a Freddy.RPC.Client process linked to the current process.

This function is used to start a Freddy.RPC.Client process in a supervision tree. The process will be started by calling init/1 with the given initial value.

Arguments

mod - the module that defines the server callbacks (like in GenServer)
connection - the pid of a Freddy.Connection process
config - the configuration of the RPC Client (describing the exchange and timeout value)
initial - the value that will be given to init/1
options - the GenServer options

Configuration

:exchange - a keyword list or %Freddy.Core.Exchange{} structure, describing an exchange that will be used to publish RPC requests to. If not present, the default RabbitMQ exchange will be used. See Freddy.Core.Exchange for available options
:timeout - specified default request timeout in milliseconds

stop(client, reason \\ :normal)

See GenServer.stop/2.

Link to this section Callbacks

before_request(request, state)

Specs

before_request(request(), state()) ::
  {:ok, state()}
  | {:ok, request(), state()}
  | {:reply, response(), state()}
  | {:stop, reason :: term(), response(), state()}
  | {:stop, reason :: term(), state()}

Called before a request will be performed to the exchange.

It receives as argument the RPC request structure which contains the message payload, the routing key and the options for that publication, and the internal client state.

Returning {:ok, state} will cause the request to be performed with no modification, block the client until the response is received, and enter the main loop with the given state.

Returning {:ok, request, state} will cause the payload, routing key and options from the given request to be used instead of the original ones, block the client until the response is received, and enter the main loop with the given state.

Returning {:reply, response, state} will respond the client inmediately without performing the request with the given response, and enter the main loop again with the given state.

Returning {:stop, reason, response, state} will not send the message, respond to the caller with response, terminate the main loop and call terminate/2 before the process exits with reason reason.

Returning {:stop, reason, state} will not send the message, terminate the main loop and call terminate/2 before the process exits with reason reason.

decode_response(payload, meta, request, state)

Specs

decode_response(payload :: String.t(), meta(), request(), state()) ::
  {:ok, payload(), state()}
  | {:ok, payload(), meta(), state()}
  | {:reply, reply :: term(), state()}
  | {:reply, reply :: term(), state(), timeout() | :hibernate}
  | {:noreply, state()}
  | {:stop, reason :: term(), state()}

Called when a response message is delivered from the queue before passing it into a on_response function.

The arguments are the message's raw payload, response metatdata, original RPC request for which the response has arrived, and the internal state.

The metadata is a map containing all metadata given by the AMQP client when receiving the message plus the :exchange and :queue values.

Returning {:ok, payload, state} or {:ok, payload, meta, state} will pass the decoded payload and meta into handle_message/3 function.

Returning {:noreply, state} will do nothing, and therefore the message should be acknowledged by using Freddy.Consumer.ack/2, Freddy.Consumer.nack/2 or Freddy.Consumer.reject/2.

Returning {:stop, reason, state} will terminate the main loop and call terminate/2 before the process exits with reason reason.

encode_request(request, state)

Specs

encode_request(request(), state()) ::
  {:ok, request(), state()}
  | {:reply, response(), state()}
  | {:reply, response(), state(), timeout() | :hibernate}
  | {:stop, reason :: term(), response(), state()}
  | {:stop, reason :: term(), state()}

Called before a message will be published to the exchange.

It receives as argument the RPC request structure and the internal state.

Returning {:ok, request, state} will cause the returned request to be published to the exchange, and the process to enter the main loop with the given state.

Returning {:reply, response, state} will respond the client inmediately without performing the request with the given response, and enter the main loop again with the given state.

Returning {:stop, reason, response, state} will not send the message, respond to the caller with response, and terminate the main loop and call terminate/2 before the process exits with reason reason.

Returning {:stop, reason, state} will not send the message, terminate the main loop and call terminate/2 before the process exits with reason reason.

handle_call(request, arg2, state)

Specs

handle_call(request :: term(), GenServer.from(), state()) ::
  {:reply, reply :: term(), state()}
  | {:reply, reply :: term(), state(), timeout() | :hibernate}
  | {:noreply, state()}
  | {:noreply, state(), timeout() | :hibernate}
  | {:stop, reason :: term(), state()}
  | {:stop, reason :: term(), reply :: term(), state()}

handle_cast(request, state)

Specs

handle_cast(request :: term(), state()) ::
  {:noreply, state()}
  | {:noreply, state(), timeout() | :hibernate}
  | {:stop, reason :: term(), state()}

Called when the process receives a cast message sent by cast/2. This callback has the same arguments as the GenServer equivalent and the :noreply and :stop return tuples behave the same.

handle_connected(arg1, state)

Specs

handle_connected(Freddy.Consumer.connection_info(), state()) ::
  {:noreply, state()}
  | {:noreply, state(), timeout() | :hibernate}
  | {:error, state()}
  | {:stop, reason :: term(), state()}

Called when the RPC client process has opened AMQP channel before registering itself as a consumer.

First argument is a map, containing :channel, :exchange and :queue structures.

Returning {:noreply, state} will cause the process to enter the main loop with the given state.

Returning {:error, state} will cause the process to reconnect (i.e. open new channel, declare exchange and queue, etc).

Returning {:stop, reason, state} will terminate the main loop and call terminate/2 before the process exits with reason reason.

handle_disconnected(reason, state)

Specs

handle_disconnected(reason :: term(), state()) ::
  {:noreply, state()} | {:stop, reason :: term(), state()}

Called when the AMQP server has been disconnected from the AMQP broker.

Returning {:noreply, state} will cause the process to enter the main loop with the given state. The server will not consume any new messages until connection to AMQP broker is restored.

Returning {:stop, reason, state} will terminate the main loop and call terminate/2 before the process exits with reason reason.

handle_info(message, state)

Specs

handle_info(message :: term(), state()) ::
  {:noreply, state()}
  | {:noreply, state(), timeout() | :hibernate}
  | {:stop, reason :: term(), state()}

Called when the process receives a message. This callback has the same arguments as the GenServer equivalent and the :noreply and :stop return tuples behave the same.

handle_ready(meta, state)

Specs

handle_ready(meta(), state()) ::
  {:noreply, state()}
  | {:noreply, state(), timeout() | :hibernate}
  | {:stop, reason :: term(), state()}

Called when the AMQP server has registered the process as a consumer of the server-named queue and it will start to receive messages.

Returning {:noreply, state} will causes the process to enter the main loop with the given state.

Returning {:stop, reason, state} will not send the message, terminate the main loop and call terminate/2 before the process exits with reason reason.

init(initial)

Specs

init(initial :: term()) :: {:ok, state()} | :ignore | {:stop, reason :: term()}

Called when the RPC client process is first started. start_link/5 will block until it returns.

It receives as argument the fourth argument given to start_link/5.

Returning {:ok, state} will cause start_link/5 to return {:ok, pid} and attempt to open a channel on the given connection, declare the exchange, declare a server-named queue, and consume it. After that it will enter the main loop with state as its internal state.

Returning :ignore will cause start_link/5 to return :ignore and the process will exit normally without entering the loop, opening a channel or calling terminate/2.

Returning {:stop, reason} will cause start_link/5 to return {:error, reason} and the process will exit with reason reason without entering the loop, opening a channel, or calling terminate/2.

on_response(response, request, state)

Specs

on_response(response(), request(), state()) ::
  {:reply, response(), state()}
  | {:reply, response(), state(), timeout() | :hibernate}
  | {:noreply, state()}
  | {:noreply, state(), timeout() | :hibernate}
  | {:stop, reason :: term(), response(), state()}
  | {:stop, reason :: term(), state()}

Called when a response has been received, before it is delivered to the caller.

It receives as argument the decoded and parse response, original RPC request for which the response has arrived, and the internal state.

Returning {:reply, reply, state} will cause the given reply to be delivered to the caller instead of the original response, and enter the main loop with the given state.

Returning {:noreply, state} will enter the main loop with the given state without responding to the caller (that will eventually timeout or keep blocked forever if the timeout was set to :infinity).

Returning {:stop, reason, reply, state} will deliver the given reply to the caller instead of the original response and call terminate/2 before the process exits with reason reason.

Returning {:stop, reason, state} not reply to the caller and call terminate/2 before the process exits with reason reason.

on_return(request, state)

Specs

on_return(request(), state()) ::
  {:reply, response(), state()}
  | {:reply, response(), state(), timeout() | :hibernate}
  | {:noreply, state()}
  | {:noreply, state(), timeout() | :hibernate}
  | {:stop, reason :: term(), response(), state()}
  | {:stop, reason :: term(), state()}

Called when a request has been returned by AMPQ broker.

Returning {:reply, reply, state} will cause the given reply to be delivered to the caller, and enter the main loop with the given state.

Returning {:stop, reason, reply, state} will deliver the given reply to the caller, and call terminate/2 before the process exits with reason reason.

Returning {:stop, reason, state} will not reply to the caller and call terminate/2 before the process exits with reason reason.

on_timeout(request, state)

Specs

on_timeout(request(), state()) ::
  {:reply, response(), state()}
  | {:reply, response(), state(), timeout() | :hibernate}
  | {:noreply, state()}
  | {:noreply, state(), timeout() | :hibernate}
  | {:stop, reason :: term(), response(), state()}
  | {:stop, reason :: term(), state()}

Called when a request has timed out.

Returning {:reply, reply, state} will cause the given reply to be delivered to the caller, and enter the main loop with the given state.

Returning {:stop, reason, reply, state} will deliver the given reply to the caller, and call terminate/2 before the process exits with reason reason.

Returning {:stop, reason, state} will not reply to the caller and call terminate/2 before the process exits with reason reason.

terminate(reason, state)

Specs

terminate(reason :: term(), state()) :: any()

This callback is the same as the GenServer equivalent and is called when the process terminates. The first argument is the reason the process is about to exit with.