ProtonStream provides a streaming API for OS processes that mirrors Elixir's Port module.

This library is derived from MuonTrap by Frank Hunleth, with a new streaming API and bidirectional stdin/stdout/stderr support.

ProtonStream can be used in two modes:

1. Message-based mode - via open/3, where the calling process receives messages
2. Callback module mode - via start_link/4, where a GenServer module handles events

Message-based Mode

Use open/3 to start a process. The calling process becomes the "owner" and receives messages about stdout, stderr, and process termination.

Messages TO ProtonStream (from owner)

• {pid, {:command, binary}} - send data to child's stdin
• {pid, :close} - close the port
• {pid, {:connect, new_pid}} - transfer ownership to another process

Messages FROM ProtonStream (to owner)

• {pid, {:data, data}} - stdout from child process
• {pid, {:error, data}} - stderr from child process
• {pid, :closed} - reply to close request
• {pid, :connected} - reply to connect request
• {:EXIT, pid, reason} - process termination (when trapping exits)

Example

{:ok, ps} = ProtonStream.open("cat", [])
send(ps, {self(), {:command, "hello"}})

receive do
  {^ps, {:data, data}} -> IO.puts("Got: #{data}")
end

send(ps, {self(), :close})

Callback Module Mode

For integration with supervision trees, use start_link/4 with a callback module that implements both GenServer and the ProtonStream behaviour.

Callbacks

The following callbacks must be implemented:

• GenServer.init/1 - called when the process starts, returns initial state
• handle_stdout/2 - called when stdout data is received
• handle_stderr/2 - called when stderr data is received
• handle_exit/2 - called when the child process exits

Example

defmodule MyWorker do
  use GenServer
  @behaviour ProtonStream

  def start_link(args) do
    ProtonStream.start_link(__MODULE__, "my_command", [], args)
  end

  @impl GenServer
  def init(args) do
    {:ok, %{buffer: "", args: args}}
  end

  @impl ProtonStream
  def handle_stdout(data, state) do
    {:noreply, %{state | buffer: state.buffer <> data}}
  end

  @impl ProtonStream
  def handle_stderr(data, state) do
    IO.puts(:stderr, data)
    {:noreply, state}
  end

  @impl ProtonStream
  def handle_exit(reason, state) do
    {:stop, reason, state}
  end
end

The module can then be added to a supervision tree:

children = [
  {MyWorker, [some: :args]}
]

Supervisor.start_link(children, strategy: :one_for_one)

Callback Return Values

• GenServer.init/1 returns {:ok, state} or {:stop, reason}
• handle_stdout/2 returns {:noreply, state} or {:stop, reason, state}
• handle_stderr/2 returns {:noreply, state} or {:stop, reason, state}
• handle_exit/2 returns {:stop, reason, state}

GenServer Callbacks

The callback module may also implement GenServer.handle_call/3, GenServer.handle_cast/2, GenServer.handle_info/2, GenServer.handle_continue/2, and GenServer.terminate/2 to handle synchronous requests, asynchronous requests, messages, continuations, and cleanup.

Configuring cgroups

On most Linux distributions, use cgcreate to create a new cgroup:

sudo cgcreate -a $(whoami) -g memory,cpu:proton_stream

Then use the :cgroup_controllers and :cgroup_base options.