Belt

v0.5.1

Belt v0.5.1 Belt.Job View Source

A mechanism for maintaining state across the Belt processing chain.

Belt.Job is implemented on top of GenServer and serves as a backchannel for Belt’s GenStage-based one-directional architecture.

Newly created Jobs are automatically supervised by Belt.Job.Supervisor which is started as part of the Belt application.

Usage: 

{:ok, job} = Belt.Job.new(:some_payload)
Belt.Job.finished?(job)
#=> false
Belt.Job.finish(job, :some_reply)
#=> :ok
Belt.Job.finished?(job)
#=> true
{:ok, reply} = Belt.Job.await_and_shutdown(job)
#=> {:ok, :some_reply}

Belt.Job.new(:some_payload)
|> Belt.Job.await_and_shutdown()
#=> :timeout

Link to this section Summary

Types

name()

The Job name

t()

The Job reference

Functions

alive?(job)

Checks if the given job is still running

await(job, timeout \\ 10000)

Subscribes current process to job (using self/0) and waits for timeout milliseconds for its completion

await_and_shutdown(job, timeout \\ 10000)

Subscribes current process to job (using self/0) and waits for timeout milliseconds for its completion

child_spec(init_arg)

Returns a specification to start this module under a supervisor

finish(job, reply)

Marks the given job as finished and stores reply as the result of the Job

finished?(job)

Checks if the given job has been completed

get_payload(job)

Returns the payload of the given job

new(payload, name \\ :auto)

Creates a new Job

register_worker(job, pid)

Registers a worker process with a given job. Workers get sent an :exit signal if they are still alive when the Job terminates

shutdown(job)

Terminates the given job

subscribe(job, pid)

Subscribes a pid to messages from the given job

Link to this section Types

Link to this type name() View Source 
name() :: {:via, module(), term()}

The Job name

Link to this type t() View Source 
t() :: pid() | name()

The Job reference

Link to this section Functions

Link to this function alive?(job) View Source 
alive?(t() | term()) :: true | false

Checks if the given job is still running.

Link to this function await(job, timeout \\ 10000) View Source 
await(t() | term(), integer() | :infinity) :: {:ok, term()} | :timeout

Subscribes current process to job (using self/0) and waits for timeout milliseconds for its completion.

:infinity can be passed for timeout if no timeout is desired.

If a matching :job_finished message is received before the timeout expires, returns {:ok, reply}. Otherwise, returns :timeout.

await/2 doesn’t terminate the given job. This can be achieved by using Belt.await/2, Belt.Job.await_and_shutdown/2 or Belt.Job.shutdown/1 instead.

Link to this function await_and_shutdown(job, timeout \\ 10000) View Source 
await_and_shutdown(t() | term(), integer() | :infinity) ::
  {:ok, term()} | :timeout

Subscribes current process to job (using self/0) and waits for timeout milliseconds for its completion.

:infinity can be passed for timeout if no timeout is desired.

If a matching :job_finished message is received before the timeout expires, returns {:ok, reply}. Otherwise, returns :timeout.

The given Job process is shut down after a matching :job_finished message has been received or timeout has expired.

Link to this function child_spec(init_arg) View Source

Returns a specification to start this module under a supervisor.

See Supervisor.

Link to this function finish(job, reply) View Source 
finish(t() | term(), term()) :: :ok | {:error, term()}

Marks the given job as finished and stores reply as the result of the Job.

All subscribers are sent the :job_finished message.

Belt.Job.finish/2 does not terminate the Job process. This can be done via Belt.Job.shutdown/1.

Link to this function finished?(job) View Source 
finished?(t() | term()) :: true | false

Checks if the given job has been completed.

Link to this function get_payload(job) View Source 
get_payload(t() | term()) :: term()

Returns the payload of the given job.

Link to this function new(payload, name \\ :auto) View Source 
new(term(), :auto | term()) :: {:ok, t()}

Creates a new Job.

If name is provided, the given term will be used for registering the new Job in Belt.Job.Registry. By default, or when :auto is passed as name, a unique name is automatically generated.

Newly created Jobs will be supervised by Belt.Job.Supervisor using a :transient restart strategy.

Link to this function register_worker(job, pid) View Source 
register_worker(t() | term(), pid()) :: :ok

Registers a worker process with a given job. Workers get sent an :exit signal if they are still alive when the Job terminates.

Link to this function shutdown(job) View Source 
shutdown(t() | term()) :: :ok

Terminates the given job.

Link to this function subscribe(job, pid) View Source 
subscribe(t() | term(), pid()) :: :ok

Subscribes a pid to messages from the given job.