Elixir：如何跨多个管道维护 Stream 状态？

Question

我的目标：我想读取文件的第一行，在管道中重组输入，然后在另一个管道中处理文件的其余部分。

我的问题：流在每个新管道上重置。

示例代码:

defmodule StrangeStream do
  fs = File.stream!("file.txt")

  Stream.take(fs, 1) |> Enum.to_list() |> IO.inspect()

  Stream.take(fs, 1) |> Enum.to_list() |> IO.inspect()
end

文本文件file.txt:

First line.
Second line.
Third line.

输出:

["First line.\n"]
["First line.\n"]

如您所见，流在每个管道中重置。每个管道都从文件的第一行开始。如何在管道调用之间保持流的状态？提前致谢！

Answer 1

TL;DR: 你不能。

elixir 中没有可变状态，因此无法维护资源状态。

唯一类似的是 suspend 减少过程中的可枚举，但即使这样也无法直接使用流。

您可以诉诸 Stream.transform/4 并自行维护状态，相应地选择管道。

旁注： Enum.to_list/1 已经终止了流，因此问题中的方法根本不起作用。

Answer 2

下面是我如何获得我想要的效果。希望它能帮助其他研究此问题的人。

再次感谢 Aleksei 为我节省了这么多时间。

defmodule StrangeStream do
  do_stuff = fn(something) ->
    # We'd do something useful here
    something
  end

  {:ok, file} = File.open("file.txt", [:read, :line])

  # Read the first line
  first_line = IO.read(file, :line)
  |>  String.trim()
  |>  do_stuff.()
  |>  IO.inspect([label: "first_line"])

  # Create side-effect streams
  print_stream = IO.binstream(:stdio, :line)
  file_stream  = File.stream!("output.txt", [:write, :append])

  # Convert IO to Stream and process
  IO.stream(file, :line)
  |>  Stream.map(&String.trim(&1))
  |>  do_stuff.()
  |>  Stream.into(print_stream, fn(s)-> s <>"\n" end)
  |>  do_stuff.()
  |>  Stream.into(file_stream)
  |>  do_stuff.()
  |>  Enum.to_list()
  |>  IO.inspect([label: "rest of file"])
end

输出

first_line: "First line."
Second line.
Third line.
rest of file: ["Second line.", "Third line."]

Elixir：如何跨多个管道维护 Stream 状态？

Elixir: how to maintain Stream state across multiple pipes?

stream

elixir