1.2.0

What's Changed

• Added Pipeline::buffer(), which provides control of the number of values buffered by the pipeline before back-pressure is applied to the data source by @trowski in #21

Full Changelog: v1.1.0...v1.2.0

1.1.0

• Added Pipeline::merge() which combines multiple iterators, emitting a value whenever any iterator emits a value.
• Fixed Pipeline::take() not completing until a value beyond the given count was emitted. The pipeline now completes immediately after emitting the last view.
• Marked the template type of ConcurrentIterator as covariant.

1.0.0

Initial stable release 🎉

Changes from 1.0.0 Beta 7

• Marked ConcurrentArrayIterator, ConcurrentChainedIterator, and ConcurrentIterableIterator as @internal. Instead of these classes, use Pipeline::fromIterable() or Pipeline::concat()
• Pipeline::concat() now accepts an array of any iterable, not only other Pipeline objects

1.0.0 Beta 7

• Removed failing a Queue that is destructed without being completed. PHP's random destruct order sometimes will lead to the Queue destructor being invoked before another destructor that would have completed the queue.

1.0.0 Beta 6

• Add compatibility with Revolt v1.x
• Improve ConcurrentIterableIterator

1.0.0 Beta 5

• Added isComplete() to the ConcurrentIterator interface that returns true when the iterator has been completed (either successfully or with an error) and no further values are pending)
• Fixed an issue where a reference to the prior value emitted on a ConcurrentIterator was held while awaiting the next value.

1.0.0 Beta 4

• PHP 8.1 is now required.
• Fixed circular references in ConcurrentIterableIterator and ConcurrentFlatMapIterator that prevented quick garbage collection, particularly problematic with instances created from Pipeline::fromIterable() using a generator.

1.0.0 Beta 3

• Pipeline has been changed from an interface to a final class. ConcurrentIterator acts as the interface replacement
• Pipeline::pipe() has been removed in favor of operator methods directly on Pipeline, such as map() and filter()
• Emitter has been renamed to Queue
  • yield() has been renamed to push()
  • emit() has been renamed to pushAsync()
• All functions in the Amp\Pipeline have been removed.
  • fromIterable() is available as Pipeline::fromIterable()
  • concat() is now Pipeline::concat()
  • Most operators are available directly on Pipeline
• Added Pipeline::generate() that invokes a closure to create each pipeline value.

Example of using Pipeline for concurrency:

use Amp\Pipeline\Pipeline;
use function Amp\delay;

$pipeline = Pipeline::fromIterable(function (): \Generator {
    for ($i = 0; $i < 100; ++$i) {
        yield $i;
    }
});

$results = $pipeline->concurrent(10)
        ->tap(fn () => delay(\random_int(1, 10) / 10))  // Delay for 0.1 to 1 seconds, simulating I/O.
        ->map(fn (int $input): int => $input * 10)
        ->filter(fn (int $input) => $input % 3 === 0); // Filter only values divisible by 3.

foreach ($results as $value) {
    echo $value, "\n";
}

1.0.0 Beta 2

• Pipeline back-pressure has been modified to be relieved immediately when the value is consumed from the pipeline. Prior, another value had to be requested from the pipeline before back-pressure was relieved.
• Removed AsyncGenerator class. Instead, fromIterable now also accepts a Closure returning an iterable, which can be a generator function.
• concurrentOrdered has been removed and concurrentUnordered renamed to concurrent. Unfortunately, ordered iteration broke if using operators that would not always emit a value, so support has been dropped.
• Added an optional $bufferSize parameter to the Emitter constructor that sets a number of items that can be emitted before awaiting back-pressure from the pipeline consumer. This value defaults to 0, which will await back-pressure with every emitted value.

1.0.0 Beta 1

Initial beta release!