Packages
livebook
0.1.1
0.19.8
0.19.7
0.19.6
0.19.5
0.19.4
0.19.3
0.19.2
0.19.1
0.19.0
0.18.6
0.18.5
0.18.4
0.18.3
0.18.2
0.18.1
0.18.0
0.17.3
0.17.2
0.17.1
0.17.0
0.16.4
0.16.3
0.16.2
0.16.1
0.16.0
0.15.5
0.15.4
0.15.3
0.15.2
0.15.1
0.15.0
0.14.7
0.14.6
0.14.5
0.14.4
0.14.3
0.14.2
0.14.1
0.14.0
0.14.0-rc.1
0.14.0-rc.0
0.13.3
0.13.2
0.13.1
0.13.0
0.12.1
0.12.0
0.11.4
0.11.3
0.11.2
0.11.1
0.11.0
0.10.0
0.9.3
0.9.2
0.9.1
0.9.0
0.8.2
0.8.1
0.8.0
0.7.2
0.7.1
0.7.0
0.6.3
0.6.2
0.6.1
0.6.0
0.5.2
0.5.1
0.5.0
0.4.1
0.4.0
0.3.2
0.3.1
0.3.0
0.2.3
0.2.2
0.2.1
0.2.0
0.1.2
0.1.1
0.1.0
Automate code & data workflows with interactive notebooks
Current section
Files
Jump to
Current section
Files
lib/livebook/utils/emitter.ex
defmodule Livebook.Utils.Emitter do
@moduledoc false
# A wrapper struct for sending messages to the specified process.
defstruct [:target_pid, :ref, :mapper]
@type t :: %__MODULE__{
target_pid: pid(),
ref: reference(),
mapper: mapper()
}
@type mapper :: (term() -> term())
@doc """
Builds a new structure where `target_pid` represents
the process that will receive all emitted items.
"""
@spec new(pid()) :: t()
def new(target_pid) do
%__MODULE__{target_pid: target_pid, ref: make_ref(), mapper: &Function.identity/1}
end
@doc """
Sends {:emitter, ref, item} message to the `target_pid`.
Note that item may be transformed with emitter's `mapper`
if there is one, see `Emitter.mapper/2`.
"""
@spec emit(t(), term()) :: :ok
def emit(emitter, item) do
message = {:emitter, emitter.ref, emitter.mapper.(item)}
send(emitter.target_pid, message)
:ok
end
@doc """
Returns a new emitter that maps all emitted items with `mapper`.
"""
@spec mapper(t(), mapper()) :: t()
def mapper(emitter, mapper) do
mapper = fn x -> mapper.(emitter.mapper.(x)) end
%{emitter | mapper: mapper}
end
end
defimpl Collectable, for: Livebook.Utils.Emitter do
alias Livebook.Utils.Emitter
def into(emitter) do
collector_fun = fn
:ok, {:cont, item} -> Emitter.emit(emitter, item)
:ok, _ -> :ok
end
{:ok, collector_fun}
end
end