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 - A struct that implements the CSP.Buffer protocol. This library ships with three implementations: CSP.Buffer.Blocking, CSP.Buffer.Dropping and CSP.Buffer.Sliding. Check out their documentation for more information on how each one works.

Using as a collection

The CSP.Channel struct has underpinnings for working alongside the Stream and Enum modules. You can use a channel directly as a Collectable:

channel = Channel.new()
Enum.into([:some, :data], channel)

channel = Channel.new()
for x <- 1..4, into: channel do
  x * x
end

And you can also use the channel as an Enumerable:

channel = Channel.new()
Enum.take(channel, 2)

channel = channel.new()
for x <- channel do
  x * x
end

Just be mindful that just like the put and get operations can be blocking, so do these. One trick is to spin up a process to take care of feeding and reading the channel:

channel = Channel.new()
spawn_link(fn -> Enum.into([:some, :data], channel) end)
Enum.take(channel, 2) # => [:some, :data]

In the next section we will discuss some of the gotchas of using channels as Collectables/Enumerables.

As a Collectable

Every element that is fed into the channel causes a putoperation, so be sure that there will be someone reading from your channel or that your channel has a buffer big enough to accommodate the incoming events.

As an Enumerable

Every element that is read from the channel causes a get operation. so be sure that there will be someone adding values to your channel.

A thing to keep in mind while using channels as Enumerables is that they act like infinite streams, so eager functions that consume the whole stream (like Enum.map/2 or Enum.each/2) will only return when the channel is closed (see CSP.Channel.close/1).

Unless that is exactly what you want, use functions from the Stream module to build your processing pipeline and finish them with something like Enum.take/2 or Enum.take_while/2.

Another caveat of using channels as Enumerables is that filtering with Stream.filter/2 or Enum.filter/2 does not simply filter what you are going to read from the channel. It reads the values and then discards the rejected ones.

If you just want to direct the filtered results elsewhere or partition the events between multiple consumers, use CSP.Channel.partition/2 and CSP.Channel.partition/3 respectively.

The almost same thing would happen with Enum.count/1 and Enum.member?/2, as the values would be read from the channel to get the result and be discarded afterwards. As an operation like this should not be done on a channel, an error is raised if those functions (and others similar to them) are called with a channel.

Using the the Collectable/Enumerable implementation you can get some nice results, like this parallel map over a channel implementation:

defmodule ChannelExtensions do
  alias CSP.Channel

  # Receive a channel, the number of workers
  # and a function to be called on each value.
  # Returns a channel with the results.
  def pmap(source, workers, fun) do
    # Create the results channel and a temporary channel just for cleanup.
    results = Channel.new()
    done = Channel.new()

    # Spin up the number of workers passed as argument.
    # Each worker stream values from on channel
    # to the other passing each to the function.
    # After the source is depleted, each worker puts
    # a message in the done channel.
    for _ <- 1..workers do
      Task.start_link(fn ->
        for value <- source, into: results, do: fun.(value)

        Channel.put(done, true)
      end)
    end

    # Spin up a cleanup process that blocks until all
    # the workers put a message in the done channel,
    # then closes both the done and the results channels.
    Task.start_link(fn ->
      Enum.take(done, workers)

      Channel.close(done)
      Channel.close(results)
    end)

    results
  end
end

OTP Compatibility

Since channels are just GenServers, you can use a channel in a supervision tree:

alias CSP.{
  Buffer,
  Channel
}

children = [
  {Channel, name: MyApp.Channel, buffer: Buffer.Blocking.new(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 as a channel struct just call CSP.Channel.wrap/1:

channel = Channel.wrap(MyApp.Channel)
Enum.into(1..4, channel)