defmodule CSP.Channel do @moduledoc """ Module used to create and manage Channels ## Options There are some options that may be used to change a channel behavior, but the channel's options can only be set during it's creation. The available options are: * `name` - Registers the channel proccess with a name. Note that the naming constraints are the same applied to a `GenServer`. * `buffer_type` - The type of the buffer used in the channel (by default `:blocking`). * `buffer_size` - The maximum capacity of the channel's buffer (by default `0`). ## Buffer types There are three suported buffer types: * `:blocking` - The channel never blocks until the buffer capacity is full. * `:sliding` - The channel never blocks, but when the buffer is full, it starts discarding the older values on it to make room for the newer ones. * `:dropping` - The channel never blocks, but when the buffer is full, it starts discarding any new values that are put on it, keeping the old ones. ## Collections interoperability You can use a channel just like any collection: channel = Channel.new other_channel = Channel.new spawn fn -> Enum.into(channel, other_channel) end Channel.put(channel, "some_data") Channel.get(other_channel) #=> "some_data" All functions from `Enum` and `Stream` are available, but you must take into consideration the blocking operations: channel = Channel.new Enum.into(1..10, channel) # This line will block until someone reads all # the ten values. ## Example An example using a channel in a supervision tree: use CSP import Supervisor.Spec children = [ worker(Channel, [[name: MyApp.Channel, buffer_size: 10]]) ] {:ok, pid} = Supervisor.start_link(children, strategy: :one_for_one) # You can use all the functions with the registered name instead # of the Channel struct Channel.put(MyApp.Channel, :data) Channel.put(MyApp.Channel, :other) Channel.get(MyApp.Channel) #=> :data Channel.get(MyApp.Channel) #=> :other # If you want to use it in a list comprehension or Enumerable function # just call `CSP.Channel.wrap/1` channel = Channel.wrap(MyApp.Channel) Enum.into(1..5, channel) Enum.count(channel) #=> 5 You can use channels in any part of list comprehensions, just remember use a buffer or run it in a separated process: channel = Enum.into(1..5, Channel.new(buffer_size: 5)) :ok = Channel.close(channel) other_channel = for x <- channel, into: Channel.new(buffer_size: 5) do x * 2 end :ok = Channel.close(other_channel) Enum.to_list(other_channel) #=> [2, 4, 6, 8, 10] """ defstruct [:ref] @default_options [buffer_size: 0, buffer_type: :blocking] @server_module CSP.Channel.Server @type options :: [option] @type option :: {:buffer_size, non_neg_integer} | {:buffer_type, buffer_type} | {:name, GenServer.name} @type buffer_type :: :blocking | :sliding | :dropping @type channel_ref :: term | t @type t :: %__MODULE__{ref: term} @doc """ Function responsible for the starting of the channel. Ideal for using a CSP in a supervision tree. """ @spec start_link(options) :: GenServer.on_start def start_link(options \\ []) do options = options ++ @default_options GenServer.start_link(@server_module, options, name: options[:name]) end @doc """ Non-linking version of `CSP.Channel.start_link/1` """ @spec start(options) :: GenServer.on_start def start(options \\ []) do options = options ++ @default_options GenServer.start(@server_module, options, name: options[:name]) end @doc """ Function responsible for creating a new channel. Useful for using channels outside of a supervision tree. ## Example iex> channel = Channel.new iex> spawn_link(fn -> Channel.put(channel, :data) end) iex> Channel.get(channel) :data """ @spec new(options) :: t def new(options \\ []) do {:ok, pid} = start_link(options) %__MODULE__{ref: pid} end @doc """ Wraps the PID or registered name in a Channel struct. If the passed in value is already a Channel struct, return it unchanged. """ @spec wrap(channel_ref) :: t def wrap(%__MODULE__{} = channel), do: channel def wrap(channel), do: %__MODULE__{ref: channel} @doc """ Function responsible for fetching a value of the channel. It will block until a value is inserted in the channel or it is closed. """ @spec get(channel_ref) :: term def get(%__MODULE__{} = channel), do: get(channel.ref) def get(channel) do GenServer.call(channel, :get, :infinity) end @doc """ Function responsible for putting a value in the channel. It may block until a value is fetched deppending on the buffer type of the channel. """ @spec put(channel_ref, term) :: :ok def put(%__MODULE__{} = channel, item), do: put(channel.ref, item) def put(channel, item) do GenServer.call(channel, {:put, item}, :infinity) end @doc """ Function responsible for closing a channel. """ @spec close(channel_ref) :: :ok def close(%__MODULE__{} = channel), do: close(channel.ref) def close(channel) do GenServer.call(channel, :close, :infinity) end @doc """ Returns `true` if the channel is closed or `false` otherwise. """ @spec closed?(channel_ref) :: boolean def closed?(%__MODULE__{} = channel), do: closed?(channel.ref) def closed?(channel) do GenServer.call(channel, :"closed?", :infinity) end @doc """ Returns the current size of the channel. """ @spec size(channel_ref) :: non_neg_integer def size(%__MODULE__{} = channel), do: size(channel.ref) def size(channel) do GenServer.call(channel, :size, :infinity) end @doc """ Returns `true` or `false` wheter the value is present on the channel. """ @spec member?(channel_ref, term) :: boolean def member?(%__MODULE__{} = channel, value), do: member?(channel.ref, value) def member?(channel, value) do GenServer.call(channel, {:"member?", value}, :infinity) end end defimpl Enumerable, for: CSP.Channel do def reduce(_channel, {:halt, acc}, _fun), do: {:halted, acc} def reduce(channel, {:suspend, acc}, fun), do: {:suspended, acc, &reduce(channel, &1, fun)} def reduce(channel, {:cont, acc}, fun) do case CSP.Channel.get(channel) do nil -> {:done, acc} value -> reduce(channel, fun.(value, acc), fun) end end def member?(channel, value), do: {:ok, CSP.Channel.member?(channel, value)} def count(channel), do: {:ok, CSP.Channel.size(channel)} end defimpl Collectable, for: CSP.Channel do def into(channel) do {channel, fn channel, {:cont, x} -> :ok = CSP.Channel.put(channel, x) channel channel, :done -> channel _, :halt -> :ok end} end end defimpl Inspect, for: CSP.Channel do import Inspect.Algebra def inspect(channel, opts) do state = cond do is_pid(channel.ref) && Process.alive?(channel.ref) && CSP.Channel.closed?(channel) -> "closed" is_pid(channel.ref) && Process.alive?(channel.ref) -> "open" :otherwise -> "not_channel" end concat ["#Channel"] end end