bufferWhile()

Buffers stream items while the predicate returns true. When the predicate evaluates to false (or the stream ends), a RunOpenCode\Component\Dataset\Model\Buffer instance is emitted and buffering restarts.

Warning

Memory usage depends on the predicate and data distribution, so memory safety depends on the specific use case.

class RunOpenCode\Component\Dataset\Operator\BufferWhile

__construct(iterable<TKey, TValue> $source, callable(Buffer<TKey, TValue>, TKey=, TValue=): bool $predicate)

Parameters:

• $source – iterable<TKey, TValue> Stream source to iterate over.

• $predicate – callable(Buffer<TKey, TValue>, TKey=, TValue=): bool Predicate function to evaluate if current item should be placed into existing buffer, or existing buffer should be yielded and new one should be created with current item.

getIterator()

Returns:

\Traversable<int, Buffer<TKey, TValue>> Stream of buffers.

Use cases

• You want to load large dataset (per example, from file or from database) and process it in batches (batch import). However, you want to cluster stream data based on yielded keys/values from data stream.

Example

Model defines Person entity and Address entity which needs to be migrated from legacy system. In the first iteration, all legacy data for Person is migrated into new system.

In the next iteration, data for Address is loaded, ordered by person_id column, and for each related Person, addresses are stored.

On each processed Person and its addresses, system is notified about successful migration of Person addresses.

<?php

use App\Model\Person;
use App\Model\Address;
use App\Events\PersonAddressesMigrated;
use RunOpenCode\Component\Dataset\Stream;
use RunOpenCode\Component\Dataset\Model\Buffer;

$orm, $dispatcher;
$dataset = $database->executeQuery('...');

new Stream($dataset)
    ->bufferWhile(function(Buffer $buffer, array $values): bool {
        return $buffer->first()->value()['person_id'] === $values['person_id'];
    })
    ->map(function(Buffer $buffer) use ($orm): Buffer {
        $person = $orm->fetchOne(Person::class, $buffer->first()->value()['person_id']);
        $addresses = new Stream($buffer)
            ->map(function(array $row): Address {
                return Address::fromLegacy($row);
            })
            ->tap(function(Address $entity) use ($orm): void {
                $orm->persist($entity);
            })
            ->collect(ArrayCollector::class);

        $person->setAddresses($addresses);

        $orm->flush();
        $orm->clear();

        return $person;
    })
    ->tap(function(Person $person) use ($dispatcher): void {
         $dispatcher->notify(new PersonAddressesMigrated($person::class, $person->id));
    })
    ->flush();