View Source ProcessHub.Service.Storage (ProcessHub v0.8.0)

Local storage service. Provides API functions for managing local storage used by ProcessHub for three concerns:

Misc storage — strategy configs, runtime state. Always ETS.
Hook storage — registered hook handlers. Always ETS.
Process registry — child registrations. Routed through the configured ProcessHub.Service.Storage.Behaviour backend (default ProcessHub.Service.Storage.Ets).

Calls to the public API that target a registry table (the hub_id atom) are dispatched through the backend of the running hub, read from its stored ProcessHub.Hub. Misc and hook storage (passed by tid) bypass the dispatcher and use ETS directly.

The public function signatures and return values are unchanged from prior releases.

Summary

Types

table_id()

A plain ETS table, a hub id (its registry backend), or an opened {module, ref} backend such as the declared list's.

Functions

clear_all(table)

Deletes all objects from the storage table.

exists?(table, key)

Returns a boolean indicating whether the key exists in local storage.

export_all(table)

Exports all objects from the storage table.

foldl(table, acc, fun)

Folds over all objects in the storage table with a function.

foldl_entries(table, acc, fun)

Folds over all objects as {key, value}, dropping the trailing TTL element that entries inserted with a :ttl carry.

get(table, key)

Returns the value for the key in local storage.

insert(table, key, value, opts \\ [])

Inserts the key and value into local storage.

insert_many(table, items, write_opts \\ [])

Inserts a batch of {key, value, opts} items.

match(table, match_expr)

Matches the given expression against the storage table.

read_durable(table)

Returns every non-expired row held in the backend's durable medium, without touching its live view. {:ok, []} for backends without one; {:error, reason} when the medium is unreadable.

registered_backend(hub_id)

Returns {module, ref} for the registry backend of the running hub hub_id, the backend itself when given one, or nil if no such hub is running.

remove(table, key, write_opts \\ [])

Removes an entry from the storage.

sync(table)

Makes every write deferred with sync: false durable. :ok for backends without a durable medium.

update(table, key, func)

Updates the value for the key in local storage.

Types

table_id()

@type table_id() :: atom() | :ets.tid() | {module(), term()}

A plain ETS table, a hub id (its registry backend), or an opened {module, ref} backend such as the declared list's.

Functions

clear_all(table)

@spec clear_all(table_id()) :: boolean() | :ok

Deletes all objects from the storage table.

Never use in custom code. Should only be used for testing purposes.

exists?(table, key)

@spec exists?(table_id(), term()) :: boolean()

Returns a boolean indicating whether the key exists in local storage.

export_all(table)

@spec export_all(table_id()) :: [{atom() | binary(), term(), pos_integer() | nil}]

Exports all objects from the storage table.

foldl(table, acc, fun)

@spec foldl(table_id(), acc, (term(), acc -> acc)) :: acc when acc: term()

Folds over all objects in the storage table with a function.

More memory efficient than export_all + filter for large tables as it doesn't create an intermediate copy of the entire table.

foldl_entries(table, acc, fun)

@spec foldl_entries(table_id(), acc, ({term(), term()}, acc -> acc)) :: acc
when acc: term()

Folds over all objects as {key, value}, dropping the trailing TTL element that entries inserted with a :ttl carry.

get(table, key)

@spec get(table_id(), term()) :: term()

Returns the value for the key in local storage.

If the key does not exist, nil is returned; otherwise, the value is returned in the format of a tuple: {key, value, ttl}.

insert(table, key, value, opts \\ [])

@spec insert(table_id(), term(), term(), keyword() | nil) :: boolean() | :ok

Inserts the key and value into local storage.

Available options:

ttl: The time to live for the key in milliseconds.

insert_many(table, items, write_opts \\ [])

@spec insert_many(table_id(), [{term(), term(), keyword()}], keyword()) ::
  :ok | {:error, term()}

Inserts a batch of {key, value, opts} items.

Backends implementing the optional insert_many/2 callback commit the whole batch with a single write, so concurrent readers never observe a partially applied batch. Other backends fall back to per-item inserts.

write_opts (sync: false) defers the backend's durable sync to a later sync/1; backends without that callback ignore it and stay durable on return.

match(table, match_expr)

@spec match(table_id(), term()) :: term()

Matches the given expression against the storage table.

The match_expr is a tuple that specifies the pattern.

read_durable(table)

@spec read_durable(table_id()) :: {:ok, [{term(), term()}]} | {:error, term()}

Returns every non-expired row held in the backend's durable medium, without touching its live view. {:ok, []} for backends without one; {:error, reason} when the medium is unreadable.

read_durable/1 is an optional backend callback: a backend that does not implement it is treated as having no durable medium.

registered_backend(hub_id)

@spec registered_backend(table_id()) :: {module(), term()} | nil

Returns {module, ref} for the registry backend of the running hub hub_id, the backend itself when given one, or nil if no such hub is running.

remove(table, key, write_opts \\ [])

@spec remove(table_id(), term(), keyword()) :: boolean() | :ok

Removes an entry from the storage.

sync(table)

@spec sync(table_id()) :: :ok | {:error, term()}

Makes every write deferred with sync: false durable. :ok for backends without a durable medium.

update(table, key, func)

@spec update(table_id(), term(), function()) :: boolean()

Updates the value for the key in local storage.

The func function is expected to take the current value as an argument and return the new value.

If the key does not exist, the current value is nil.