Limit & time budget
Stop an infinite stream cleanly
.limit(n) lets the first n items through, then stops the pipeline, which makes an infinite source (generator, itertools.count()) usable. .timeout(s) stops the step once s seconds have elapsed since it started: it is an overall time budget, not an inactivity delay between two items.
Signature
Pipeline.limit(packet_limit: int) -> Pipeline[InPacket]
Pipeline.timeout(timeout: float) -> Pipeline[InPacket]python
import itertools
import time
from olympipe import Pipeline
# An infinite source, cut after 5 items
print(Pipeline(itertools.count()).task(lambda x: x * x).limit(5).wait_for_result())
def ticker():
i = 0
while True:
yield i
i += 1
time.sleep(0.3)
# A 2-second budget on an endless stream
start = time.time()
res = Pipeline(ticker()).timeout(2.0).wait_for_result()
print(res, f"{time.time() - start:.1f}s")Output
[0, 1, 4, 9, 16]
[0, 1, 2, 3, 4, 5, 6, 7] 2.4sHow it works
When the limit or the budget is reached, the step signals the stop and the source is no longer read. Upstream steps run in parallel: the source may have produced a few more items than n, which are simply dropped. If the source is blocked (for instance in a sleep), the full stop waits until it yields control.