Поток

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

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

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

Сводка

Типы

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

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

Функции

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)

(с версии 1.5.0)

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

Сокращённая запись для chunk_every(enum, count, count).

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

(с версии 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)

(с версии 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)

(с версии 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)

(с версии 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()

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

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

Эта функция можно рассматривать как комбинацию 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)

(с версии 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}]