Потоки

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

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

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

В приведенном выше примере, когда мы применяли отображение к диапазону, элементы перечислялись по одному, во время перечисления. Модуль 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(): Индекс, отсчитываемый с нуля.

Функции

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): Лениво пропускает элементы перечисляемого объекта, пока заданная функция возвращает истинное значение.
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): Преобразует существующий поток с функциями начала и завершения.
unfold(next_acc, next_fun): Генерирует последовательность значений для данного аккумулирующего значения.
uniq(enum): Создаёт поток, который генерирует только уникальные элементы.
uniq_by(enum, fun): Создаёт поток, который генерирует только уникальные элементы, удаляя элементы, для которых функция fun возвратила дубликаты.
with_index(enum, offset \\ 0): Создаёт поток, где каждый элемент в перечисляемом объекте будет обернут в кортеж вместе с его индексом.
zip(enumerables): Сжимает соответствующие элементы из конечного набора перечисляемых объектов в один поток кортежей.
zip(left, right): Лениво сжимает два набора данных вместе.

Типы

acc()

Характеристики

acc() :: any()

default()

Характеристики

default() :: any()

element()

Характеристики

element() :: any()

index()

Характеристики

index() :: non_neg_integer()

Индекс, отсчитываемый с нуля.

END_OF_DOCUMENT_MARKER

Функции

chunk_by(enum, fun)

Спецификации

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)

Спецификации

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

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

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

Спецификации

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]]

chunk_while(enum, acc, chunk_fun, after_fun)

Спецификации

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)

Спецификации

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)

Спецификации

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)

Спецификации

cycle(Enumerable.t()) :: Enumerable.t()

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

Примеры

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

dedup(enum)

Спецификации

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)

Спецификации

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)

Спецификации

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)

Спецификации

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)

Спецификации

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]

each(enum, fun)

Спецификации

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

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

Полезно для добавления побочных эффектов (например, печати) в поток.

Примеры

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)

Спецификации

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)

Спецификации

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)

Спецификации

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)

Спецификации

interval(non_neg_integer()) :: 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(enum, collectable, transform \\ fn x -> x end)

Спецификации

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

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

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

iterate(start_value, next_fun)

Спецификации

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)

Спецификации

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)

Спецификации

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)

Спецификации

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]

END_OF_DOCUMENT_MARKER

repeatedly(generator_fun)

Характеристики

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

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

Примеры

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

resource(start_fun, next_fun, after_fun)

Характеристики

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
)

run(stream)

Характеристики

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)

Характеристики

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)

Характеристики

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)

Характеристики

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)

Характеристики

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)

Характеристики

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)

Характеристики

timer(non_neg_integer()) :: Enumerable.t()

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

Выдаваемое значение —

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

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

Примеры

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

transform(enum, acc, reducer)

Характеристики

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 = 1..100
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)
[1, 2, 3]

transform(enum, start_fun, reducer, after_fun)

Характеристики

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

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

Аккумулятор вычисляется только при запуске преобразования. Также позволяет указать функцию после завершения, которая вызывается, когда поток останавливается или завершается.

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

unfold(next_acc, next_fun)

Характеристики

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)

Характеристики

uniq(Enumerable.t()) :: Enumerable.t()

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

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

Примеры

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

uniq_by(enum, fun)

Характеристики

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)

Характеристики

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)

Характеристики

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(left, right)

Характеристики

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}]