Исходный код Поток

Функции для создания и комбинирования потоков.

Потоки — это комбинируемые, ленивые перечислимые (для ознакомления с перечислимыми, см. модуль Enum). Любой перечислимый, который генерирует элементы по одному во время перечисления, называется потоком. Например, Range в Elixir является потоком:

iex> range = 1..5
1..5
iex> Enum.map(range, &(&1 * 2))
[2, 4, 6, 8, 10]

В примере выше, при применении `map` к диапазону, элементы генерировались по одному, во время перечисления. Модуль Stream позволяет отобразить диапазон, не вызывая его перечисления:

iex> range = 1..3
iex> stream = Stream.map(range, &(&1 * 2))
iex> Enum.map(stream, &(&1 + 1))
[3, 5, 7]

Обратите внимание, что мы начали с диапазона, а затем создали поток, который предназначен для умножения каждого элемента в диапазоне на 2. На этом этапе вычисления не производились. Только при вызове функции Enum.map/2 мы фактически перечисляем каждый элемент в диапазоне, умножаем его на 2 и добавляем 1. Мы говорим, что функции в Stream являются ленивыми, а функции в Enum — жадными.

Благодаря своей ленивости, потоки полезны при работе с большими (или даже бесконечными) коллекциями. При комбинировании многих операций с Enum создаются промежуточные списки, тогда как Stream создаёт рецепт вычислений, которые выполняются в более поздний момент. Посмотрим ещё один пример:

1..3
|> Enum.map(&IO.inspect(&1))
|> Enum.map(&(&1 * 2))
|> Enum.map(&IO.inspect(&1))
1
2
3
2
4
6
#=> [2, 4, 6]

Обратите внимание, что сначала мы вывели каждый элемент в списке, затем умножили каждый элемент на 2 и, наконец, вывели каждое новое значение. В этом примере список перечислялся трижды. Посмотрим пример с потоками:

stream = 1..3
|> Stream.map(&IO.inspect(&1))
|> Stream.map(&(&1 * 2))
|> Stream.map(&IO.inspect(&1))
Enum.to_list(stream)
1
2
2
4
3
6
#=> [2, 4, 6]

Хотя конечный результат одинаковый, порядок вывода элементов изменился! С потоками мы выводим первый элемент, а затем выводим его удвоенное значение. В этом примере список перечислялся только один раз!

Вот что мы имели в виду, когда говорили ранее, что потоки являются комбинируемыми, ленивыми перечислимыми. Обратите внимание, что мы можем вызывать Stream.map/2 несколько раз, эффективно комбинируя потоки и сохраняя их ленивыми. Вычисления выполняются только при вызове функции из модуля Enum.

Как и в случае с Enum, функции в этом модуле работают за линейное время. Это означает, что время, необходимое для выполнения операции, растёт с той же скоростью, что и длина списка. Это ожидается для операций, таких как Stream.map/2. В конце концов, если мы хотим пройти по каждому элементу потока, чем длиннее поток, тем больше элементов нам нужно пройти, и тем дольше это займёт.

Создание потоков

Существует множество функций в стандартной библиотеке Elixir, которые возвращают потоки, некоторые примеры:

IO.stream/2 - потоки строк ввода, по одной
URI.query_decoder/1 - декодирует строку запроса, пара за парой

Этот модуль также предоставляет множество удобных функций для создания потоков, таких как Stream.cycle/1, Stream.unfold/2, Stream.resource/3 и другие.

Обратите внимание, что функции в этом модуле гарантированно возвращают перечислимые. Поскольку перечислимые могут иметь различные формы (структуры, анонимные функции и т. д.), функции в этом модуле могут возвращать любые из этих форм, и это может измениться в любое время. Например, функция, которая сегодня возвращает анонимную функцию, может возвращать структуру в будущих выпусках.

Резюме

Типы

acc()
default()
element()
index(): Индекс с нулевым основанием.
timer()

Функции

chunk_by(enum, fun): Разбивает enum на куски, буферизуя элементы, для которых fun возвращает одинаковое значение.
chunk_every(enum, count): Является сокращением для chunk_every(enum, count, count).
chunk_every(enum, count, step, leftover \\ []): Потоковая передача перечисляемого набора в куски, содержащие по count элементов каждый, где каждый новый кусок начинается с позиции step в перечисляемом наборе.
chunk_while(enum, acc, chunk_fun, after_fun): Разбивает enum на куски с тонким управлением, когда каждый кусок выводится.
concat(enumerables): Создаёт поток, который перечисляет каждый перечисляемый набор в перечисляемом наборе.
concat(first, second): Создаёт поток, который перечисляет первый аргумент, за которым следует второй.
cycle(enumerable): Создаёт поток, который циклически проходит по заданному перечисляемому набору бесконечно.
dedup(enum): Создаёт поток, который выводит элементы только в том случае, если они отличаются от последнего выведенного элемента.
dedup_by(enum, fun): Создаёт поток, который выводит элементы только в том случае, если результат вызова fun на элементе отличается от результата (сохранённого) вызова fun на последнем выведенном элементе.
drop(enum, n): Лениво пропускает следующие n элементов из перечисляемого набора.
drop_every(enum, nth): Создаёт поток, который пропускает каждый nth элемент из перечисляемого набора.
drop_while(enum, fun): Лениво пропускает элементы перечисляемого набора, пока заданная функция возвращает истинное значение.
duplicate(value, n): Дублирует заданный элемент n раз в потоке.
each(enum, fun): Выполняет заданную функцию для каждого элемента.
filter(enum, fun): Создаёт поток, который фильтрует элементы в соответствии с заданной функцией при перечислении.
flat_map(enum, mapper): Применяет заданную fun к enumerable и сплющивает результат.
intersperse(enumerable, intersperse_element): Лениво вставляет intersperse_element между каждым элементом перечисления.
interval(n): Создаёт поток, который выводит значение через заданный интервал n миллисекунд.
into(enum, collectable, transform \\ fn x -> x end): Вводит значения потока в заданный соберёмый набор как побочный эффект.
iterate(start_value, next_fun): Выводит последовательность значений, начиная с start_value. Последующие значения генерируются путём вызова next_fun на предыдущем значении.
map(enum, fun): Создаёт поток, который применит заданную функцию при перечислении.
map_every(enum, nth, fun): Создаёт поток, который применит заданную функцию к каждому nth элементу из перечисляемого набора.
reject(enum, fun): Создаёт поток, который отклонит элементы в соответствии с заданной функцией при перечислении.
repeatedly(generator_fun): Возвращает поток, сгенерированный путём многократного вызова generator_fun.
resource(start_fun, next_fun, after_fun): Выводит последовательность значений для данного ресурса.
run(stream): Запускает заданный поток.
scan(enum, fun): Создаёт поток, который применяет заданную функцию к каждому элементу, выводит результат и использует тот же результат в качестве аккумулятора для следующего вычисления. Использует первый элемент в перечисляемом наборе в качестве начального значения.
scan(enum, acc, fun): Создаёт поток, который применяет заданную функцию к каждому элементу, выводит результат и использует тот же результат в качестве аккумулятора для следующего вычисления. Использует заданное acc в качестве начального значения.
take(enum, count): Лениво берёт следующие count элементов из перечисляемого набора и останавливает перечисление.
take_every(enum, nth): Создаёт поток, который берёт каждый nth элемент из перечисляемого набора.
take_while(enum, fun): Лениво берёт элементы перечисляемого набора, пока заданная функция возвращает истинное значение.
timer(n): Создаёт поток, который выводит единственное значение через n миллисекунд.
transform(enum, acc, reducer): Преобразует существующий поток.
transform(enum, start_fun, reducer, after_fun): Аналогично Stream.transform/5, за исключением того, что last_fun не предоставляется.
transform(enum, start_fun, reducer, last_fun, after_fun): Преобразует существующий поток с функциями-обработчиками начала, конца и завершения.
unfold(next_acc, next_fun): Выводит последовательность значений для данного аккумулятора.
uniq(enum): Создаёт поток, который выводит только уникальные элементы.
uniq_by(enum, fun): Создаёт поток, который выводит только уникальные элементы, удаляя элементы, для которых функция fun возвратила дублируемые элементы.
with_index(enum, offset \\ 0): Создаёт поток, где каждый элемент в перечисляемом наборе будет обернут в кортеж вместе с его индексом.
zip(enumerables): Сжимает соответствующие элементы из конечного набора перечисляемых наборов в один поток кортежей.
zip(enumerable1, enumerable2): Лениво сжимает два перечисляемых набора вместе.

END_OF_DOCUMENT_MARKER

zip_with(enumerables, zip_fun): Лениво объединяет соответствующие элементы из конечного набора перечисляемых объектов в новый перечисляемый объект, преобразуя их с помощью функции zip_fun, по мере обработки.
zip_with(enumerable1, enumerable2, zip_fun): Лениво объединяет соответствующие элементы из двух перечисляемых объектов в новый, преобразуя их с помощью функции zip_fun, по мере обработки.

Типы

acc()Source

@type acc() :: any()

default()Source

@type default() :: any()

element()Source

@type element() :: any()

index()Source

@type index() :: non_neg_integer()

Индекс, нумерация с нуля.

timer()Source

@type timer() :: non_neg_integer() | :infinity

Функции

chunk_by(enum, fun)Source

@spec chunk_by(Enumerable.t(), (element() -> any())) :: Enumerable.t()

Разбивает перечислимый объект enum на части, буферизируя элементы, для которых fun возвращает одно и то же значение.

Элементы выводятся только тогда, когда fun возвращает новое значение или перечисление enum завершается.

Примеры

iex> stream = Stream.chunk_by([1, 2, 2, 3, 4, 4, 6, 7, 7], &(rem(&1, 2) == 1))
iex> Enum.to_list(stream)
[[1], [2, 2], [3], [4, 4, 6], [7, 7]]

chunk_every(enum, count)Source

@spec chunk_every(Enumerable.t(), pos_integer()) :: Enumerable.t()

Является сокращением для chunk_every(enum, count, count).

chunk_every(enum, count, step, leftover \\ [])Source

@spec chunk_every(
  Enumerable.t(),
  pos_integer(),
  pos_integer(),
  Enumerable.t() | :discard
) ::
  Enumerable.t()

Выводит перечислимый объект по частям, содержащим по count элементов в каждой, где каждая новая часть начинается step элементов в перечислимом объекте.

step является необязательным и, если не указано, по умолчанию равно count, т.е. части не перекрываются. Разбиение на части прекратится, как только закончится коллекция или будет выведена неполная часть.

Если последняя часть не имеет count элементов для заполнения части, элементы берутся из leftover для заполнения части. Если leftover не имеет достаточно элементов для заполнения части, возвращается неполная часть с менее чем count элементами.

Если :discard задано в leftover, последняя часть отбрасывается, если только она не содержит ровно count элементов.

Примеры

iex> Stream.chunk_every([1, 2, 3, 4, 5, 6], 2) |> Enum.to_list()
[[1, 2], [3, 4], [5, 6]]

iex> Stream.chunk_every([1, 2, 3, 4, 5, 6], 3, 2, :discard) |> Enum.to_list()
[[1, 2, 3], [3, 4, 5]]

iex> Stream.chunk_every([1, 2, 3, 4, 5, 6], 3, 2, [7]) |> Enum.to_list()
[[1, 2, 3], [3, 4, 5], [5, 6, 7]]

iex> Stream.chunk_every([1, 2, 3, 4, 5, 6], 3, 3, []) |> Enum.to_list()
[[1, 2, 3], [4, 5, 6]]

iex> Stream.chunk_every([1, 2, 3, 4], 3, 3, Stream.cycle([0])) |> Enum.to_list()
[[1, 2, 3], [4, 0, 0]]

chunk_while(enum, acc, chunk_fun, after_fun)Source

@spec chunk_while(
  Enumerable.t(),
  acc(),
  (element(), acc() -> {:cont, chunk, acc()} | {:cont, acc()} | {:halt, acc()}),
  (acc() -> {:cont, chunk, acc()} | {:cont, acc()})
) :: Enumerable.t()
when chunk: any()

Разбивает перечислимый объект enum с тонкими настройками, когда каждая часть передаётся.

chunk_fun получает текущий элемент и накопитель и должен вернуть {:cont, element, acc} для вывода данной части и продолжения с накопителем или {:cont, acc} для того, чтобы не выводить никакую часть и продолжить с возвращаемым накопителем.

after_fun вызывается, когда итерация завершена, и также должна вернуть {:cont, element, acc} или {:cont, acc}.

Примеры

iex> chunk_fun = fn element, acc ->
...>   if rem(element, 2) == 0 do
...>     {:cont, Enum.reverse([element | acc]), []}
...>   else
...>     {:cont, [element | acc]}
...>   end
...> end
iex> after_fun = fn
...>   [] -> {:cont, []}
...>   acc -> {:cont, Enum.reverse(acc), []}
...> end
iex> stream = Stream.chunk_while(1..10, [], chunk_fun, after_fun)
iex> Enum.to_list(stream)
[[1, 2], [3, 4], [5, 6], [7, 8], [9, 10]]

concat(enumerables)Source

@spec concat(Enumerable.t()) :: Enumerable.t()

Создаёт поток, который перечисляет каждый перечислимый объект в перечислимом объекте.

Примеры

iex> stream = Stream.concat([1..3, 4..6, 7..9])
iex> Enum.to_list(stream)
[1, 2, 3, 4, 5, 6, 7, 8, 9]

concat(first, second)Source

@spec concat(Enumerable.t(), Enumerable.t()) :: Enumerable.t()

Создаёт поток, который перечисляет первый аргумент, а затем второй.

Примеры

iex> stream = Stream.concat(1..3, 4..6)
iex> Enum.to_list(stream)
[1, 2, 3, 4, 5, 6]

iex> stream1 = Stream.cycle([1, 2, 3])
iex> stream2 = Stream.cycle([4, 5, 6])
iex> stream = Stream.concat(stream1, stream2)
iex> Enum.take(stream, 6)
[1, 2, 3, 1, 2, 3]

cycle(enumerable)Source

@spec cycle(Enumerable.t()) :: Enumerable.t()

Создаёт поток, который циклически перебирает заданный перечислимый объект бесконечно.

Примеры

iex> stream = Stream.cycle([1, 2, 3])
iex> Enum.take(stream, 5)
[1, 2, 3, 1, 2]

dedup(enum)Source

@spec dedup(Enumerable.t()) :: Enumerable.t()

Создаёт поток, который выводит элементы только если они отличаются от последнего выведенного элемента.

Эта функция всегда должна хранить только последний выведенный элемент.

Элементы сравниваются с помощью ===/2.

Примеры

iex> Stream.dedup([1, 2, 3, 3, 2, 1]) |> Enum.to_list()
[1, 2, 3, 2, 1]

dedup_by(enum, fun)Source

@spec dedup_by(Enumerable.t(), (element() -> term())) :: Enumerable.t()

Создаёт поток, который выводит элементы только если результат вызова fun на элементе отличается от результата вызова fun на последнем выведенном элементе.

Примеры

iex> Stream.dedup_by([{1, :x}, {2, :y}, {2, :z}, {1, :x}], fn {x, _} -> x end) |> Enum.to_list()
[{1, :x}, {2, :y}, {1, :x}]

drop(enum, n)Source

@spec drop(Enumerable.t(), integer()) :: Enumerable.t()

Лениво отбрасывает следующие n элементов из перечислимого объекта.

Если задан отрицательный n, отбрасываются последние n элементы из коллекции. Обратите внимание, что механизм, с помощью которого это реализовано, откладывает вывод любого элемента до тех пор, пока не будет выведено n дополнительных элементов перечислением.

Примеры

iex> stream = Stream.drop(1..10, 5)
iex> Enum.to_list(stream)
[6, 7, 8, 9, 10]

iex> stream = Stream.drop(1..10, -5)
iex> Enum.to_list(stream)
[1, 2, 3, 4, 5]

drop_every(enum, nth)Source

@spec drop_every(Enumerable.t(), non_neg_integer()) :: Enumerable.t()

Создаёт поток, который отбрасывает каждый nth элемент из перечислимого объекта.

Первый элемент всегда отбрасывается, если nth не равно 0.

nth должно быть неотрицательным целым числом.

Примеры

iex> stream = Stream.drop_every(1..10, 2)
iex> Enum.to_list(stream)
[2, 4, 6, 8, 10]

iex> stream = Stream.drop_every(1..1000, 1)
iex> Enum.to_list(stream)
[]

iex> stream = Stream.drop_every([1, 2, 3, 4, 5], 0)
iex> Enum.to_list(stream)
[1, 2, 3, 4, 5]

drop_while(enum, fun)Source

@spec drop_while(Enumerable.t(), (element() -> as_boolean(term()))) :: Enumerable.t()

Лениво отбрасывает элементы перечислимого объекта, пока заданная функция возвращает истинное значение.

Примеры

iex> stream = Stream.drop_while(1..10, &(&1 <= 5))
iex> Enum.to_list(stream)
[6, 7, 8, 9, 10]

duplicate(value, n)Source

@spec duplicate(any(), non_neg_integer()) :: Enumerable.t()

Дублирует заданный элемент n раз в потоке.

n — целое число, большее или равное 0.

Если n равно 0, возвращается пустой поток.

Примеры

iex> stream = Stream.duplicate("hello", 0)
iex> Enum.to_list(stream)
[]

iex> stream = Stream.duplicate("hi", 1)
iex> Enum.to_list(stream)
["hi"]

iex> stream = Stream.duplicate("bye", 2)
iex> Enum.to_list(stream)
["bye", "bye"]

iex> stream = Stream.duplicate([1, 2], 3)
iex> Enum.to_list(stream)
[[1, 2], [1, 2], [1, 2]]

each(enum, fun)Source

@spec each(Enumerable.t(), (element() -> term())) :: Enumerable.t()

Выполняет заданную функцию для каждого элемента.

Значения в потоке не изменяются, поэтому эта функция полезна для добавления побочных эффектов (например, вывода на печать) в поток. См. map/2, если нужно создать другой поток.

Примеры

iex> stream = Stream.each([1, 2, 3], fn x -> send(self(), x) end)
iex> Enum.to_list(stream)
iex> receive do: (x when is_integer(x) -> x)
1
iex> receive do: (x when is_integer(x) -> x)
2
iex> receive do: (x when is_integer(x) -> x)
3

filter(enum, fun)Source

@spec filter(Enumerable.t(), (element() -> as_boolean(term()))) :: Enumerable.t()

Создаёт поток, фильтрующий элементы согласно заданной функции при перечислении.

Примеры

iex> stream = Stream.filter([1, 2, 3], fn x -> rem(x, 2) == 0 end)
iex> Enum.to_list(stream)
[2]

flat_map(enum, mapper)Source

@spec flat_map(Enumerable.t(), (element() -> Enumerable.t())) :: Enumerable.t()

Применяет заданную функцию fun к enumerable и уплощает результат.

Эта функция возвращает новый поток, полученный путём добавления результата вызова fun для каждого элемента enumerable вместе.

Примеры

iex> stream = Stream.flat_map([1, 2, 3], fn x -> [x, x * 2] end)
iex> Enum.to_list(stream)
[1, 2, 2, 4, 3, 6]

iex> stream = Stream.flat_map([1, 2, 3], fn x -> [[x]] end)
iex> Enum.to_list(stream)
[[1], [2], [3]]

intersperse(enumerable, intersperse_element)Source

@spec intersperse(Enumerable.t(), any()) :: Enumerable.t()

Лениво вставляет intersperse_element между каждым элементом перечисления.

Примеры

iex> Stream.intersperse([1, 2, 3], 0) |> Enum.to_list()
[1, 0, 2, 0, 3]

iex> Stream.intersperse([1], 0) |> Enum.to_list()
[1]

iex> Stream.intersperse([], 0) |> Enum.to_list()
[]

interval(n)Source

@spec interval(timer()) :: Enumerable.t()

Создаёт поток, который выводит значение через заданный интервал n миллисекунд.

Выводимые значения — это возрастающий счётчик, начинающийся с 0. Эта операция блокирует вызывающий процесс заданным интервалом каждый раз, когда передаётся новый элемент.

Не используйте эту функцию для генерации последовательности чисел. Если блокирование вызывающего процесса не требуется, используйте Stream.iterate(0, & &1 + 1) вместо этого.

Примеры

iex> Stream.interval(10) |> Enum.take(10)
[0, 1, 2, 3, 4, 5, 6, 7, 8, 9]

END_OF_DOCUMENT_MARKER

into(enum, collectable, transform \\ fn x -> x end)Source

@spec into(Enumerable.t(), Collectable.t(), (term() -> term())) :: Enumerable.t()

Вставляет значения потока в задаваемый коллектор как побочный эффект.

Эта функция часто используется с run/1, так как любые вычисления отложены до выполнения потока. См. run/1 для примера.

iterate(start_value, next_fun)Source

@spec iterate(element(), (element() -> element())) :: Enumerable.t()

Выдает последовательность значений, начиная с start_value. Последующие значения генерируются путём вызова next_fun над предыдущим значением.

Примеры

iex> Stream.iterate(0, &(&1 + 1)) |> Enum.take(5)
[0, 1, 2, 3, 4]

map(enum, fun)Source

@spec map(Enumerable.t(), (element() -> any())) :: Enumerable.t()

Создаёт поток, который будет применять данную функцию к перечислению.

Примеры

iex> stream = Stream.map([1, 2, 3], fn x -> x * 2 end)
iex> Enum.to_list(stream)
[2, 4, 6]

map_every(enum, nth, fun)Source

@spec map_every(Enumerable.t(), non_neg_integer(), (element() -> any())) ::
  Enumerable.t()

Создаёт поток, который будет применять данную функцию к каждому nth элементу из перечислимого объекта.

Первый элемент всегда передаётся в данную функцию.

nth должно быть неотрицательным целым числом.

Примеры

iex> stream = Stream.map_every(1..10, 2, fn x -> x * 2 end)
iex> Enum.to_list(stream)
[2, 2, 6, 4, 10, 6, 14, 8, 18, 10]

iex> stream = Stream.map_every([1, 2, 3, 4, 5], 1, fn x -> x * 2 end)
iex> Enum.to_list(stream)
[2, 4, 6, 8, 10]

iex> stream = Stream.map_every(1..5, 0, fn x -> x * 2 end)
iex> Enum.to_list(stream)
[1, 2, 3, 4, 5]

reject(enum, fun)Source

@spec reject(Enumerable.t(), (element() -> as_boolean(term()))) :: Enumerable.t()

Создаёт поток, который будет отбрасывать элементы в соответствии с данной функцией на перечислении.

Примеры

iex> stream = Stream.reject([1, 2, 3], fn x -> rem(x, 2) == 0 end)
iex> Enum.to_list(stream)
[1, 3]

repeatedly(generator_fun)Source

@spec repeatedly((-> element())) :: Enumerable.t()

Возвращает поток, сгенерированный путём вызова generator_fun повторно.

Примеры

# Although not necessary, let's seed the random algorithm
iex> :rand.seed(:exsss, {1, 2, 3})
iex> Stream.repeatedly(&:rand.uniform/0) |> Enum.take(3)
[0.5455598952593053, 0.6039309974353404, 0.6684893034823949]

resource(start_fun, next_fun, after_fun)Source

@spec resource(
  (-> acc()),
  (acc() -> {[element()], acc()} | {:halt, acc()}),
  (acc() -> term())
) ::
  Enumerable.t()

Выдает последовательность значений для данного ресурса.

Аналогично transform/3, но начальное накопленное значение вычисляется лениво через start_fun и выполняет after_fun в конце перечисления (в обоих случаях успеха и неудачи).

Последовательные значения генерируются путём вызова next_fun с предыдущим аккумулятором (начальное значение — результат, возвращаемый start_fun) и он должен вернуть кортеж, содержащий список элементов для вывода и следующий аккумулятор. Перечисление завершается, если оно возвращает {:halt, acc}.

Как следует из названия, эта функция полезна для потоковой передачи значений из ресурсов.

Примеры

Stream.resource(
  fn -> File.open!("sample") end,
  fn file ->
    case IO.read(file, :line) do
      data when is_binary(data) -> {[data], file}
      _ -> {:halt, file}
    end
  end,
  fn file -> File.close(file) end
)

iex> Stream.resource(
...>  fn ->
...>    {:ok, pid} = StringIO.open("string")
...>    pid
...>  end,
...>  fn pid ->
...>    case IO.getn(pid, "", 1) do
...>      :eof -> {:halt, pid}
...>      char -> {[char], pid}
...>    end
...>  end,
...>  fn pid -> StringIO.close(pid) end
...> ) |> Enum.to_list()
["s", "t", "r", "i", "n", "g"]

run(stream)Source

@spec run(Enumerable.t()) :: :ok

Выполняет данный поток.

Это полезно, когда поток нужно выполнить для побочных эффектов, и его результат не интересует.

Примеры

Открываем файл, заменяем все # на % и передаём в другой файл без загрузки всего файла в память:

File.stream!("/path/to/file")
|> Stream.map(&String.replace(&1, "#", "%"))
|> Stream.into(File.stream!("/path/to/other/file"))
|> Stream.run()

Никаких вычислений не будет произведено, пока мы не вызовем одну из функций Enum или run/1.

scan(enum, fun)Source

@spec scan(Enumerable.t(), (element(), acc() -> any())) :: Enumerable.t()

Создаёт поток, который применяет данную функцию к каждому элементу, выводит результат и использует тот же результат в качестве аккумулятора для следующего вычисления. Использует первый элемент в перечислимом объекте в качестве начального значения.

Примеры

iex> stream = Stream.scan(1..5, &(&1 + &2))
iex> Enum.to_list(stream)
[1, 3, 6, 10, 15]

scan(enum, acc, fun)Source

@spec scan(Enumerable.t(), acc(), (element(), acc() -> any())) :: Enumerable.t()

Создаёт поток, который применяет данную функцию к каждому элементу, выводит результат и использует тот же результат в качестве аккумулятора для следующего вычисления. Использует заданное acc в качестве начального значения.

Примеры

iex> stream = Stream.scan(1..5, 0, &(&1 + &2))
iex> Enum.to_list(stream)
[1, 3, 6, 10, 15]

take(enum, count)Source

@spec take(Enumerable.t(), integer()) :: Enumerable.t()

Лениво берёт следующие count элементы из перечислимого объекта и останавливает перечисление.

Если задано отрицательное count, последние count значения будут взяты. В таком случае коллекция полностью перечисляется, сохраняя до 2 * count элементов в памяти. После достижения конца коллекции, последние count элементы будут выполнены. Поэтому использование отрицательного count на бесконечном наборе никогда не вернёт результат.

Примеры

iex> stream = Stream.take(1..100, 5)
iex> Enum.to_list(stream)
[1, 2, 3, 4, 5]

iex> stream = Stream.take(1..100, -5)
iex> Enum.to_list(stream)
[96, 97, 98, 99, 100]

iex> stream = Stream.cycle([1, 2, 3]) |> Stream.take(5)
iex> Enum.to_list(stream)
[1, 2, 3, 1, 2]

take_every(enum, nth)Source

@spec take_every(Enumerable.t(), non_neg_integer()) :: Enumerable.t()

Создаёт поток, который берёт каждый nth элемент из перечислимого объекта.

Первый элемент всегда включается, если nth не равно 0.

nth должно быть неотрицательным целым числом.

Примеры

iex> stream = Stream.take_every(1..10, 2)
iex> Enum.to_list(stream)
[1, 3, 5, 7, 9]

iex> stream = Stream.take_every([1, 2, 3, 4, 5], 1)
iex> Enum.to_list(stream)
[1, 2, 3, 4, 5]

iex> stream = Stream.take_every(1..1000, 0)
iex> Enum.to_list(stream)
[]

take_while(enum, fun)Source

@spec take_while(Enumerable.t(), (element() -> as_boolean(term()))) :: Enumerable.t()

Лениво берёт элементы из перечислимого объекта, пока данная функция возвращает истинное значение.

Примеры

iex> stream = Stream.take_while(1..100, &(&1 <= 5))
iex> Enum.to_list(stream)
[1, 2, 3, 4, 5]

timer(n)Source

@spec timer(timer()) :: Enumerable.t()

Создаёт поток, который выводит единственное значение через n миллисекунд.

Выводимое значение равно 0. Эта операция заблокирует вызывающую функцию на заданное время до тех пор, пока элемент не будет передан.

Примеры

iex> Stream.timer(10) |> Enum.to_list()
[0]

transform(enum, acc, reducer)Source

@spec transform(Enumerable.t(), acc, fun) :: Enumerable.t()
when fun: (element(), acc -> {Enumerable.t(), acc} | {:halt, acc}), acc: any()

Преобразует существующий поток.

Ожидает аккумулятора и функции, которая получает два аргумента — элемент потока и обновлённый аккумулятор. Она должна вернуть кортеж, где первый элемент — новый поток (часто список) или атом :halt, а второй элемент — аккумулятор, который будет использоваться следующим элементом.

Примечание: эта функция эквивалентна Enum.flat_map_reduce/3, за исключением того, что эта функция не возвращает аккумулятор после обработки потока.

Примеры

Stream.transform/3 полезна, так как может быть использована как основа для реализации многих функций, определённых в этом модуле. Например, мы можем реализовать Stream.take(enum, n) следующим образом:

iex> enum = 1001..9999
iex> n = 3
iex> stream = Stream.transform(enum, 0, fn i, acc ->
...>   if acc < n, do: {[i], acc + 1}, else: {:halt, acc}
...> end)
iex> Enum.to_list(stream)
[1001, 1002, 1003]

Stream.transform/5 ещё более обобщает эту функцию, позволяя оборачивать ресурсы.

transform(enum, start_fun, reducer, after_fun)Source

@spec transform(Enumerable.t(), start_fun, reducer, after_fun) :: Enumerable.t()
when start_fun: (-> acc),
     reducer: (element(), acc -> {Enumerable.t(), acc} | {:halt, acc}),
     after_fun: (acc -> term()),
     acc: any()

Аналогично Stream.transform/5, за исключением того, что last_fun не предоставляется.

Эта функция может быть рассмотрена как сочетание Stream.resource/3 с Stream.transform/3.

END_OF_DOCUMENT_MARKER

transform(enum, start_fun, reducer, last_fun, after_fun)Source

@spec transform(Enumerable.t(), start_fun, reducer, last_fun, after_fun) ::
  Enumerable.t()
when start_fun: (-> acc),
     reducer: (element(), acc -> {Enumerable.t(), acc} | {:halt, acc}),
     last_fun: (acc -> {Enumerable.t(), acc} | {:halt, acc}),
     after_fun: (acc -> term()),
     acc: any()

Преобразует существующий поток с функциями обратного вызова начала, конца и после.

После начала преобразования вызывается start_fun, чтобы вычислить начальное значение аккумулятора. Затем для каждого элемента в перечислимом объекте вызывается функция reducer, которая принимает элемент и аккумулятор, возвращая новые элементы и новый аккумулятор, как в transform/3.

После обработки коллекции вызывается last_fun, чтобы вывести все оставшиеся элементы. Затем вызывается after_fun, чтобы закрыть любой ресурс, но не выводить новые элементы. last_fun вызывается только в том случае, если перечислимый объект завершился успешно (либо потому, что он закончен, либо он сам остановился). after_fun всегда вызывается, поэтому after_fun необходимо использовать для закрытия ресурсов.

unfold(next_acc, next_fun)Source

@spec unfold(acc(), (acc() -> {element(), acc()} | nil)) :: Enumerable.t()

Выводит последовательность значений для данного аккумулятора.

Последовательные значения генерируются путем вызова next_fun с предыдущим аккумулятором, и он должен вернуть кортеж с текущим значением и следующим аккумулятором. Перечисление завершается, если он вернёт nil.

Примеры

Для создания потока, который считает вниз и останавливается перед нулём:

iex> Stream.unfold(5, fn
...>   0 -> nil
...>   n -> {n, n - 1}
...> end) |> Enum.to_list()
[5, 4, 3, 2, 1]

Если next_fun никогда не вернёт nil, возвращаемый поток является бесконечным:

iex> Stream.unfold(0, fn
...>   n -> {n, n + 1}
...> end) |> Enum.take(10)
[0, 1, 2, 3, 4, 5, 6, 7, 8, 9]

iex> Stream.unfold(1, fn
...>   n -> {n, n * 2}
...> end) |> Enum.take(10)
[1, 2, 4, 8, 16, 32, 64, 128, 256, 512]

uniq(enum)Source

@spec uniq(Enumerable.t()) :: Enumerable.t()

Создаёт поток, который выводит только уникальные элементы.

Обратите внимание, что для определения уникальности элемента функция должна хранить все уникальные значения, выводимые потоком. Поэтому, если поток бесконечен, количество хранящихся элементов будет расти до бесконечности, не подвергаясь сборке мусора.

Примеры

iex> Stream.uniq([1, 2, 3, 3, 2, 1]) |> Enum.to_list()
[1, 2, 3]

uniq_by(enum, fun)Source

@spec uniq_by(Enumerable.t(), (element() -> term())) :: Enumerable.t()

Создаёт поток, который выводит только уникальные элементы, удаляя элементы, для которых функция fun возвращает дублирующиеся элементы.

Функция fun отображает каждый элемент в терм, который используется для определения, являются ли два элемента дубликатами.

Пример

iex> Stream.uniq_by([{1, :x}, {2, :y}, {1, :z}], fn {x, _} -> x end) |> Enum.to_list()
[{1, :x}, {2, :y}]

iex> Stream.uniq_by([a: {:tea, 2}, b: {:tea, 2}, c: {:coffee, 1}], fn {_, y} -> y end) |> Enum.to_list()
[a: {:tea, 2}, c: {:coffee, 1}]

with_index(enum, offset \\ 0)Source

@spec with_index(Enumerable.t(), integer()) :: Enumerable.t()

Создаёт поток, где каждый элемент в перечислимом объекте будет заключён в кортеж вместе со своим индексом.

Если задан offset, индексация будет начинаться с указанного смещения вместо нуля.

Примеры

iex> stream = Stream.with_index([1, 2, 3])
iex> Enum.to_list(stream)
[{1, 0}, {2, 1}, {3, 2}]

iex> stream = Stream.with_index([1, 2, 3], 3)
iex> Enum.to_list(stream)
[{1, 3}, {2, 4}, {3, 5}]

zip(enumerables)Source

@spec zip(enumerables) :: Enumerable.t()
when enumerables: [Enumerable.t()] | Enumerable.t()

Сцепляет соответствующие элементы из конечного набора перечислимых объектов в один поток кортежей.

Сцепление заканчивается, как только любой перечислимый объект в заданном наборе завершается.

Примеры

iex> concat = Stream.concat(1..3, 4..6)
iex> cycle = Stream.cycle(["foo", "bar", "baz"])
iex> Stream.zip([concat, [:a, :b, :c], cycle]) |> Enum.to_list()
[{1, :a, "foo"}, {2, :b, "bar"}, {3, :c, "baz"}]

zip(enumerable1, enumerable2)Source

@spec zip(Enumerable.t(), Enumerable.t()) :: Enumerable.t()

Сцепляет два перечислимых объекта вместе лениво.

Сцепление заканчивается, как только завершается любой из перечислимых объектов.

Примеры

iex> concat = Stream.concat(1..3, 4..6)
iex> cycle = Stream.cycle([:a, :b, :c])
iex> Stream.zip(concat, cycle) |> Enum.to_list()
[{1, :a}, {2, :b}, {3, :c}, {4, :a}, {5, :b}, {6, :c}]

zip_with(enumerables, zip_fun)Source

@spec zip_with(enumerables, (Enumerable.t() -> term())) :: Enumerable.t()
when enumerables: [Enumerable.t()] | Enumerable.t()

Лениво сцепляет соответствующие элементы из конечного набора перечислимых объектов в новый перечислимый объект, преобразуя их с помощью функции zip_fun по ходу выполнения.

Первый элемент из каждого перечислимого объекта в enumerables помещается в список, который затем передаётся в одноаргументную функцию zip_fun. Затем, второй элемент из каждого перечислимого объекта помещается в список и передаётся в zip_fun, и так далее, до тех пор, пока любой из перечислимых объектов в enumerables не завершит работу.

Возвращает новый перечислимый объект с результатами вызова zip_fun.

Примеры

iex> concat = Stream.concat(1..3, 4..6)
iex> Stream.zip_with([concat, concat], fn [a, b] -> a + b end) |> Enum.to_list()
[2, 4, 6, 8, 10, 12]

iex> concat = Stream.concat(1..3, 4..6)
iex> Stream.zip_with([concat, concat, 1..3], fn [a, b, c] -> a + b + c end) |> Enum.to_list()
[3, 6, 9]

zip_with(enumerable1, enumerable2, zip_fun)Source

@spec zip_with(Enumerable.t(), Enumerable.t(), (term(), term() -> term())) ::
  Enumerable.t()

Лениво сцепляет соответствующие элементы из двух перечислимых объектов в новый, преобразуя их с помощью функции zip_fun по ходу выполнения.

Функция zip_fun будет вызываться с первым элементом из enumerable1 и первым элементом из enumerable2, затем со вторым элементом каждого и так далее, до тех пор, пока один из перечислимых объектов не завершит работу.

Примеры

iex> concat = Stream.concat(1..3, 4..6)
iex> Stream.zip_with(concat, concat, fn a, b -> a + b end) |> Enum.to_list()
[2, 4, 6, 8, 10, 12]