defmodule Actors.Actor.CallerProducer do
@moduledoc """
# Actors.Actor.CallerProducer
This module defines a GenStage producer responsible for managing actor-related
operations such as retrieving actor state, actor registration, actor spawning, and actor invocation.
## Client API
The client API provides functions for interacting with actors.
These functions can be used to initiate operations such as getting actor state,
registering actors, spawning new actors, and invoking actors.
### Functions
- [`start_link/1`](#start_link-1): Starts the CallerProducer process.
- [`get_state/2`](#get_state-2): Retrieves the state of a specified actor.
- [`register/2`](#register-2): Registers an actor with the specified registration request.
- [`spawn_actor/2`](#spawn_actor-2): Spawns an actor based on the specified spawn request.
- [`invoke/2`](#invoke-2): Invokes an actor with the specified invocation request.
- [`enqueue/1`](#enqueue-1): Enqueues an event for processing.
## Server API
The server API includes functions that handle the GenStage behavior.
These functions manage the state and processing of events.
### Functions
- [`init/1`](#init-1): Initializes the GenStage process.
- [`handle_demand/2`](#handle_demand-2): Handles demand requests from consumers.
- [`handle_call/3`](#handle_call-3): Handles call requests from consumers.
- [`handle_cast/2`](#handle_cast-2): Handles cast requests from consumers.
## Usage
To interact with this module, use the client API functions provided.
These functions handle backpressure if enabled in the configuration.
"""
use GenStage
require Logger
alias Actors.Actor.CallerConsumer
alias Actors.Config.PersistentTermConfig, as: Config
alias Eigr.Functions.Protocol.Actors.ActorId
alias Eigr.Functions.Protocol.{
InvocationRequest,
RegistrationRequest,
RegistrationResponse,
SpawnRequest,
SpawnResponse
}
# Client API
@doc """
Starts the CallerProducer process.
## Parameters
- `state` (any): Initial state for the process.
## Returns
- `:ignore`: If the process is already running.
- `{:error, reason}`: If an error occurs during process initialization.
- `{:ok, pid}`: If the process starts successfully.
"""
@spec start_link(any) :: :ignore | {:error, any} | {:ok, pid}
def start_link(state \\ []) do
GenStage.start_link(__MODULE__, state, name: __MODULE__)
end
@doc """
Retrieves the state of a specified actor.
## Parameters
- `actor_id` (ActorId.t()): The ID of the actor.
- `opts` (any): Additional options.
## Returns
- `{:ok, state}`: If the state is successfully retrieved.
- `{:error, reason}`: If an error occurs during the operation.
"""
@spec get_state(ActorId.t()) :: {:ok, term()} | {:error, term()}
def get_state(actor_id, opts \\ []) do
if Config.get(:actors_global_backpressure_enabled) do
GenStage.call(__MODULE__, {:enqueue, {:get_state, actor_id, opts}}, :infinity)
else
CallerConsumer.get_state(actor_id)
end
end
@doc """
Performs a readiness check for a given actor.
## Parameters
- `actor_id` (ActorId.t()): The ID of the actor.
- `opts` (any): Additional options.
## Returns
- `{:ok, response}`: If the response is successfully.
- `{:error, reason}`: If an error occurs during the operation.
"""
@spec readiness(ActorId.t()) :: {:ok, term()} | {:error, term()}
def readiness(actor_id, opts \\ []) do
if Config.get(:actors_global_backpressure_enabled) do
GenStage.call(__MODULE__, {:enqueue, {:readiness, actor_id, opts}}, :infinity)
else
CallerConsumer.readiness(actor_id)
end
end
@doc """
Performs a liveness check for a given actor.
## Parameters
- `actor_id` (ActorId.t()): The ID of the actor.
- `opts` (any): Additional options.
## Returns
- `{:ok, response}`: If the response is successfully.
- `{:error, reason}`: If an error occurs during the operation.
"""
@spec liveness(ActorId.t()) :: {:ok, term()} | {:error, term()}
def liveness(actor_id, opts \\ []) do
if Config.get(:actors_global_backpressure_enabled) do
GenStage.call(__MODULE__, {:enqueue, {:liveness, actor_id, opts}}, :infinity)
else
CallerConsumer.liveness(actor_id)
end
end
@doc """
Registers an actor with the specified registration request.
## Parameters
- `registration` (RegistrationRequest.t()): The registration request.
- `opts` (any): Additional options.
## Returns
- `{:ok, response}`: If the actor is successfully registered.
- `{:error, response}`: If an error occurs during the registration.
"""
@spec register(RegistrationRequest.t(), any()) ::
{:ok, RegistrationResponse.t()} | {:error, RegistrationResponse.t()}
def register(registration, opts \\ []) do
if Config.get(:actors_global_backpressure_enabled) do
GenStage.call(__MODULE__, {:enqueue, {:register, registration, opts}}, :infinity)
else
CallerConsumer.register(registration, opts)
end
end
@doc """
Spawns an actor based on the specified spawn request.
## Parameters
- `spawn_req` (SpawnRequest.t()): The spawn request.
- `opts` (any): Additional options.
## Returns
- `{:ok, response}`: If the actor is successfully spawned.
- `{:error, response}`: If an error occurs during the spawning.
"""
@spec spawn_actor(SpawnRequest.t(), any()) :: {:ok, SpawnResponse.t()}
def spawn_actor(spawn_req, opts \\ []) do
if Config.get(:actors_global_backpressure_enabled) do
GenStage.call(__MODULE__, {:enqueue, {:spawn_actor, spawn_req, opts}}, :infinity)
else
CallerConsumer.spawn_actor(spawn_req, opts)
end
end
@doc """
Invokes an actor with the specified invocation request.
## Parameters
- `request` (InvocationRequest.t()): The invocation request.
- `opts` (any): Additional options.
## Returns
- `{:ok, :async}`: If the invocation is asynchronous.
- `{:ok, result}`: If the invocation is successful.
- `{:error, reason}`: If an error occurs during the invocation.
"""
@spec invoke(InvocationRequest.t()) :: {:ok, :async} | {:ok, term()} | {:error, term()}
def invoke(request, opts \\ [])
def invoke(%InvocationRequest{} = request, opts) do
if Sidecar.GracefulShutdown.get_status() in [:draining, :stopping] do
raise ErlangError, "The ActorHost is shutting down and can no longer receive invocations"
end
if Config.get(:actors_global_backpressure_enabled) do
if request.async do
GenStage.cast(__MODULE__, {:enqueue, {:invoke, request, opts}})
{:ok, :async}
else
GenStage.call(__MODULE__, {:enqueue, {:invoke, request, opts}}, :infinity)
end
else
if request.register_ref != "" and not is_nil(request.register_ref) do
spawn_req = %SpawnRequest{
actors: [%ActorId{request.actor.id | parent: request.register_ref}]
}
spawn_actor(spawn_req, opts)
end
CallerConsumer.invoke_with_span(request, opts)
end
end
@doc """
Enqueues an event for processing.
## Parameters
- `event` (any): The event to be enqueued.
"""
def enqueue(event) do
GenStage.call(__MODULE__, {:enqueue, event})
end
# Server API
@doc false
def init(_state) do
{:producer, {:queue.new(), 0}}
end
@doc false
def handle_demand(incoming_demand, {queue, pending_demand}) do
Logger.debug("Consumer pull demand of: #{incoming_demand} elements.")
dispatch_events(queue, incoming_demand + pending_demand, [])
end
@doc false
def handle_call({:enqueue, event}, from, {queue, pending_demand}) do
queue = :queue.in({from, event}, queue)
dispatch_events(queue, pending_demand, [])
end
@doc false
def handle_cast({:enqueue, event}, {queue, pending_demand}) do
queue = :queue.in({:fake_from, event}, queue)
dispatch_events(queue, pending_demand, [])
end
defp dispatch_events(queue, 0, events) do
{:noreply, Enum.reverse(events), {queue, 0}}
end
defp dispatch_events(queue, demand, events) do
case :queue.out(queue) do
{{:value, {from, event}}, queue} ->
dispatch_events(queue, demand - 1, [{from, event} | events])
{:empty, queue} ->
{:noreply, Enum.reverse(events), {queue, demand}}
end
end
end
