Поток

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

Потоки — это композируемые, ленивые перечисляемые (для знакомства с перечисляемыми, см. модуль 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.

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

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

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

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

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

Сводка

Функции

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()

Функции

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) (since 1.5.0)

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

Сокращение для chunk_every(enum, count, count).

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

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) (since 1.5.0)

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 item, acc ->
...>   if rem(item, 2) == 0 do
...>     {:cont, Enum.reverse([item | acc]), []}
...>   else
...>     {:cont, [item | 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(), non_neg_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) (since 1.6.0)

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) (since 1.4.0)

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]

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 миллисекунд.

Излучаемое значение равно 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) (since 1.4.0)

zip(Enumerable.t()) :: Enumerable.t()

zip([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}]