Stream

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

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

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

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

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): Лениво комбинирует два перечислимых объекта.
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]

into(перечисление, коллектор, преобразование \\ fn x -> x end)Source

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

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

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

iterate(начальное_значение, функция_следующего)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(перечисление, функция)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(перечисление, n-й, функция)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(перечисление, функция)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(генератор_функция)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(функция_начала, функция_следующего, функция_после)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(поток)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(перечисление, функция)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(перечисление, накопитель, функция)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(перечисление, количество)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(перечисление, n-й)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(перечисление, функция)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(перечисление, накопитель, редуктор)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(перечисление, функция_начала, редуктор, функция_после)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]

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]