如何在Elixir中用Port实现类Bash管道的外部进程交互？

问题 #

想要在Elixir中实现与echo foo | nkf等效的功能：调用外部进程nkf，向其STDIN发送数据，从STDOUT获取完整转换结果，并确认处理完成。但直接用Port实现时，发送的单条数据会被拆分成多条消息返回，无法判断数据结束时机（示例中"foo"实际对应完整文件内容）：

iex(1)> port = Port.open({:spawn, "nkf -u"}, [:binary])
#Port<0.7>
iex(2)> Port.command(port, "foo")
true
iex(3)> flush
{#Port<0.7>, {:data, "fo"}}
{#Port<0.7>, {:data, "o"}}
:ok

如何用Elixir的Port实现与Bash管道相同的行为？

解决方案 #

核心思路是：发送完所有输入后关闭Port的STDIN，让外部进程nkf知晓输入已结束，待其处理完数据并退出后，通过退出消息确认处理完成，同时收集所有返回的数据流。

实现代码 #

defmodule Nkf do
  def run(input) do
    # 打开Port时添加:exit_status选项，用于接收进程退出状态
    port = Port.open({:spawn, "nkf -u"}, [:binary, :exit_status])
    # 向nkf发送输入数据
    Port.command(port, input)
    # 关闭Port的STDIN，告知nkf没有更多输入
    Port.close(port)
    # 递归收集所有输出，直到收到退出状态
    collect_output(port, "")
  end

  defp collect_output(port, accumulated_output) do
    receive do
      # 收集单次返回的数据，追加到结果中
      {^port, {:data, chunk}} ->
        collect_output(port, accumulated_output <> chunk)
      # 进程正常退出，返回完整结果
      {^port, {:exit_status, 0}} ->
        {:ok, accumulated_output}
      # 进程异常退出，返回错误状态和已收集的输出
      {^port, {:exit_status, status}} ->
        {:error, {:non_zero_exit, status, accumulated_output}}
    end
  end
end

使用示例 #

iex> Nkf.run("foo")
{:ok, "foo"}

# 若处理大文件，直接传入文件内容即可
iex> File.read!("large_file.txt") |> Nkf.run()
{:ok, "转换后的完整内容"}

关键说明 #

1. :exit_status选项：打开Port时添加该选项，外部进程退出时会发送{:exit_status, status}消息，以此作为处理完成的标志。
2. 关闭STDIN：调用Port.close(port)会关闭Port的输入流，nkf收到EOF后会处理剩余数据并退出，避免进程一直等待输入。
3. 收集所有数据流：通过递归的receive块，将每次返回的:data片段拼接成完整结果，直到收到退出状态消息，确保不会遗漏任何数据。

内容的提问来源于stack exchange，提问作者Adam Millerchip