Packages

spawn

1.0.0-rc.21
2.0.0-RC9 2.0.0-RC8 2.0.0-RC7 2.0.0-RC6 2.0.0-RC5 2.0.0-RC4 2.0.0-RC3 2.0.0-RC2 2.0.0-RC14 2.0.0-RC13 2.0.0-RC12 2.0.0-RC11 2.0.0-RC10 2.0.0-RC1 1.4.3 1.4.2 1.4.1 1.4.0 1.3.3 1.3.2 1.3.1 1.3.0 1.2.1 1.2.0 1.1.1 1.1.0 1.0.1 1.0.0 1.0.0-rc3 1.0.0-rc16 1.0.0-rc1 1.0.0-rc.38 1.0.0-rc.37 1.0.0-rc.36 1.0.0-rc.35 1.0.0-rc.34 1.0.0-rc.33 1.0.0-rc.32 1.0.0-rc.31 1.0.0-rc.30 1.0.0-rc.29 1.0.0-rc.28 1.0.0-rc.27 1.0.0-rc.26 1.0.0-rc.25 1.0.0-rc.24 1.0.0-rc.23 1.0.0-rc.22 1.0.0-rc.21 1.0.0-rc.20 1.0.0-rc.19 1.0.0-rc.18 1.0.0-rc.17 1.0.0-rc.2 0.6.3 0.6.2 0.6.1 0.6.0 0.5.5 0.5.4 0.5.3 0.5.1 0.5.0 0.5.0-rc.13 0.5.0-rc.12 0.5.0-rc.11 0.5.0-rc.10 0.5.0-rc.9 0.5.0-rc.8 0.5.0-rc.7 0.5.0-rc.6 0.5.0-rc.5 0.5.0-rc.3 0.5.0-alpha.13 0.5.0-alpha.12 0.5.0-alpha.11 0.5.0-alpha.10 0.5.0-alpha.9 0.5.0-alpha.8 0.5.0-alpha.7 0.5.0-alpha.6 0.5.0-alpha.5 0.5.0-alpha.4 0.5.0-alpha.3 0.5.0-alpha.2 0.5.0-alpha.1 0.1.0

Spawn is the core lib for Spawn Actors System

HexDocs

Current section

Files

Jump to
Readme
87 Versions
35 Dependencies
Dependants
Files
Activity
spawn lib actors actor caller_producer.ex
Raw

lib/actors/actor/caller_producer.ex 

defmodule Actors.Actor.CallerProducer do

  use GenStage

  require Logger



  alias Eigr.Functions.Protocol.Actors.ActorId



  alias Eigr.Functions.Protocol.{

    InvocationRequest,

    RegistrationRequest,

    RegistrationResponse,

    SpawnRequest,

    SpawnResponse

  }



  # Client API



  @spec start_link(any) :: :ignore | {:error, any} | {:ok, pid}

  def start_link(state \\ []) do

    GenStage.start_link(__MODULE__, state, name: __MODULE__)

  end



  @spec get_state(ActorId.t()) :: {:ok, term()} | {:error, term()}

  def get_state(actor_id, opts \\ []) do

    GenStage.call(__MODULE__, {:enqueue, {:get_state, actor_id, opts}}, :infinity)

  end



  @spec register(RegistrationRequest.t(), any()) ::

          {:ok, RegistrationResponse.t()} | {:error, RegistrationResponse.t()}

  def register(registration, opts \\ []) do

    GenStage.call(__MODULE__, {:enqueue, {:register, registration, opts}}, :infinity)

  end



  @spec spawn_actor(SpawnRequest.t(), any()) :: {:ok, SpawnResponse.t()}

  def spawn_actor(spawn_req, opts \\ []) do

    GenStage.call(__MODULE__, {:enqueue, {:spawn_actor, spawn_req, opts}}, :infinity)

  end



  @spec invoke(InvocationRequest.t()) :: {:ok, :async} | {:ok, term()} | {:error, term()}

  def invoke(request, opts \\ []) do

    GenStage.call(__MODULE__, {:enqueue, {:invoke, request, opts}}, :infinity)

  end



  def enqueue(event) do

    GenStage.call(__MODULE__, {:enqueue, event})

  end



  # Server API



  def init(_state) do

    {:producer, {:queue.new(), 0}}

  end



  def handle_call({:enqueue, event}, from, {queue, pending_demand}) do

    queue = :queue.in({from, event}, queue)

    dispatch_events(queue, pending_demand, [])

  end



  def handle_demand(incoming_demand, {queue, pending_demand}) do

    Logger.debug("Producer Handle Demand: #{incoming_demand}.")

    dispatch_events(queue, incoming_demand + 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