Потоки

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

Потоки — это комбинируемые, ленивые перечислимые (для ознакомления с перечислимыми, см. модуль 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(): Индекс, начинающийся с нуля.
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): Лениво пропускает элементы перечислимого объекта, пока заданная функция возвращает истинное значение.
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()

Specs

acc() :: any()

default()

Specs

default() :: any()

element()

Specs

element() :: any()

index()

Specs

index() :: non_neg_integer()

Индекс, начинающийся с нуля.

timer()

Specs

timer() :: non_neg_integer() | :infinity

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(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(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(:exrop, {1, 2, 3})
iex> Stream.repeatedly(&:rand.uniform/0) |> Enum.take(3)
[0.7498295129076106, 0.06161655489244533, 0.7924073127680873]

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
)

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

run(stream)

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

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(timer()) :: 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 = 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]

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