Поток

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

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

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 и другие.

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

Резюме

Типы

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)

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

each(enum, fun)

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

filter(enum, fun)

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

flat_map(enum, mapper)

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

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)

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

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)

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

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

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

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

Примеры

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

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

Примеры

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

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]

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

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

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

Примеры

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

stream = File.stream!("code")
|> Stream.map(&String.replace(&1, "#", "%"))
|> Stream.into(File.stream!("new"))
|> Stream.run

Никакие вычисления не будут выполнены, пока мы не вызовем одну из функций Enum или Stream.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(перечислитель, количество)

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

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(перечислитель, функция)

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

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

Примеры

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(перечислитель, накопитель, редуктор)

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(перечислитель, функция_начала, редуктор, функция_окончания)

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(следующий_накопитель, следующая_функция)

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(перечислитель)

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

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

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

Примеры

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

uniq_by(перечислитель, функция)

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(перечислитель, смещение \\ 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(перечислители)

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(левый, правый)

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