Proposal for buffer operators

Author: TheCelavi

src/RunOpenCode/Component/Dataset/src/Collector/IterableCollector.php

@@ -28,7 +28,7 @@ final class IterableCollector implements \IteratorAggregate, CollectorInterface
             return $this->getIterator();
         }
     }
-    
+
     /**
      * {@inheritdoc}
      */
@@ -45,7 +45,7 @@ final class IterableCollector implements \IteratorAggregate, CollectorInterface
 
     /**
      * Provides you with total number of iterated elements.
-     * 
+     *
      * @var non-negative-int
      */
     public private(set) int $count = 0;
@@ -73,4 +73,4 @@ public function getIterator(): \Traversable
             $this->count++;
         }
     }
-}
+}

src/RunOpenCode/Component/Dataset/src/Model/Buffer.php

@@ -0,0 +1,94 @@
+<?php
+
+declare(strict_types=1);
+
+namespace RunOpenCode\Component\Dataset\Model;
+
+/**
+ * Buffer of iterated items from collection.
+ *
+ * @template-covariant TKey
+ * @template-covariant TValue
+ *
+ * @phpstan-type ItemTuple = array{TKey, TValue}
+ *
+ * @implements \IteratorAggregate<TKey, TValue>
+ */
+final class Buffer implements \IteratorAggregate, \Countable
+{
+    /**
+     * Create new buffer.
+     *
+     * @param non-empty-list<ItemTuple> $items Items within buffer.
+     *
+     * @internal
+     */
+    public function __construct(private array &$items)
+    {
+        // noop.
+    }
+
+    /**
+     * Get first item in buffer.
+     *
+     * @return Item<TKey, TValue>
+     *
+     * @phpstan-ignore-next-line generics.variance
+     */
+    public function first(): Item
+    {
+        return new Item($this->items[0][0], $this->items[0][1]);
+    }
+
+    /**
+     * Get last item in buffer.
+     *
+     * @return Item<TKey, TValue>
+     *
+     * @phpstan-ignore-next-line generics.variance
+     */
+    public function last(): Item
+    {
+        $last = $this->items[\count($this->items) - 1];
+
+        return new Item($last[0], $last[1]);
+    }
+
+    /**
+     * Get all keys.
+     *
+     * @return list<TKey>
+     */
+    public function keys(): array
+    {
+        return \array_map(static fn(array $item): mixed => $item[0], $this->items);
+    }
+
+    /**
+     * Get all values.
+     *
+     * @return list<TValue>
+     */
+    public function values(): array
+    {
+        return \array_map(static fn(array $item): mixed => $item[1], $this->items);
+    }
+
+    /**
+     * {@inheritdoc}
+     */
+    public function getIterator(): \Traversable
+    {
+        foreach ($this->items as [$key, $value]) {
+            yield $key => $value;
+        }
+    }
+
+    /**
+     * {@inheritdoc}
+     */
+    public function count(): int
+    {
+        return \count($this->items);
+    }
+}

src/RunOpenCode/Component/Dataset/src/Model/Item.php

@@ -0,0 +1,88 @@
+<?php
+
+declare(strict_types=1);
+
+namespace RunOpenCode\Component\Dataset\Model;
+
+use RunOpenCode\Component\Dataset\Exception\OutOfBoundsException;
+
+/**
+ * @template TKey
+ * @template TValue
+ *
+ * @implements \ArrayAccess<int, TKey|TValue>
+ */
+final readonly class Item implements \ArrayAccess
+{
+    /**
+     * @param TKey   $key
+     * @param TValue $value
+     */
+    public function __construct(
+        private mixed $key,
+        private mixed $value
+    ) {
+        // noop.
+    }
+
+    /**
+     * Get key.
+     *
+     * @return TKey
+     */
+    public function key(): mixed
+    {
+        return $this->key;
+    }
+
+    /**
+     * Get value.
+     *
+     * @return TValue
+     */
+    public function value(): mixed
+    {
+        return $this->value;
+    }
+
+    /**
+     * {@inheritdoc}
+     */
+    public function offsetExists(mixed $offset): bool
+    {
+        return $offset === 0 || $offset === 1;
+    }
+
+    /**
+     * {@inheritdoc}
+     *
+     * @return ($offset is 0 ? TKey : TValue)
+     */
+    public function offsetGet(mixed $offset): mixed
+    {
+        return match ($offset) {
+            0 => $this->key,
+            1 => $this->value,
+            default => throw new OutOfBoundsException($offset, $this, \sprintf(
+                'Item tuple does not have offset "%s".',
+                $offset
+            )),
+        };
+    }
+
+    /**
+     * {@inheritdoc}
+     */
+    public function offsetSet(mixed $offset, mixed $value): void
+    {
+        throw new \BadMethodCallException('Item is immutable.');
+    }
+
+    /**
+     * {@inheritdoc}
+     */
+    public function offsetUnset(mixed $offset): void
+    {
+        throw new \BadMethodCallException('Item is immutable.');
+    }
+}

src/RunOpenCode/Component/Dataset/src/Operator/Batch.php

@@ -0,0 +1,58 @@
+<?php
+
+declare(strict_types=1);
+
+namespace RunOpenCode\Component\Dataset\Operator;
+
+use RunOpenCode\Component\Dataset\AbstractStream;
+use RunOpenCode\Component\Dataset\Contract\OperatorInterface;
+use RunOpenCode\Component\Dataset\Model\Buffer;
+
+/**
+ * Buffer count operator.
+ *
+ * Iterates over given collection and creates a buffer of items with number of items
+ * up to given capacity.
+ *
+ * Yields created instances of {@see Buffer} for batch processing.
+ *
+ * @template TKey
+ * @template TValue
+ *
+ * @extends AbstractStream<int, Buffer<TKey, TValue>>
+ * @implements OperatorInterface<int, Buffer<TKey, TValue>>
+ */
+final class BufferCount extends AbstractStream implements OperatorInterface
+{
+    /**
+     * @param iterable<TKey, TValue> $collection Collection to iterate over.
+     * @param positive-int           $count      How many items to buffer.
+     */
+    public function __construct(
+        private readonly iterable $collection,
+        private readonly int      $count = 1000,
+    ) {
+        parent::__construct($collection);
+    }
+
+    /**
+     * {@inheritdoc}
+     */
+    protected function iterate(): \Traversable
+    {
+        $items = [];
+
+        foreach ($this->collection as $key => $value) {
+            $items[] = [$key, $value];
+
+            if (\count($items) === $this->count) {
+                yield new Buffer($items);
+                $items = [];
+            }
+        }
+
+        if (\count($items) !== 0) {
+            yield new Buffer($items);
+        }
+    }
+}