Let item reader to drive item indexes

src/batch/src/Job/Item/ItemJob.php

@@ -77,17 +77,15 @@ final protected function doExecute(JobExecution $jobExecution): void
 
         $writeCount = 0;
         $itemsToWrite = [];
-        $lineNumber = 1;
-        foreach ($this->reader->read() as $readItem) {
-            $lineNumber++;
+        foreach ($this->reader->read() as $readIndex => $readItem) {
             $summary->increment('read');
 
             try {
                 $processedItem = $this->processor->process($readItem);
             } catch (InvalidItemException $exception) {
                 $summary->increment('invalid');
                 $jobExecution->addWarning(
-                    new Warning($exception->getMessage(), $exception->getParameters(), ['line_number' => $lineNumber])
+                    new Warning($exception->getMessage(), $exception->getParameters(), ['itemIndex' => $readIndex])
                 );
 
                 continue;

src/batch/src/Job/Item/Reader/IndexWithReader.php

@@ -0,0 +1,99 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Yokai\Batch\Job\Item\Reader;
+
+use Closure;
+use Yokai\Batch\Job\Item\ElementConfiguratorTrait;
+use Yokai\Batch\Job\Item\FlushableInterface;
+use Yokai\Batch\Job\Item\InitializableInterface;
+use Yokai\Batch\Job\Item\ItemReaderInterface;
+use Yokai\Batch\Job\JobExecutionAwareInterface;
+use Yokai\Batch\Job\JobExecutionAwareTrait;
+
+/**
+ * An {@see ItemReaderInterface} that decorates another {@see ItemReaderInterface}
+ * and extract item index of each item using a {@see Closure}.
+ *
+ * Provided {@see Closure} must accept a single argument (the read item)
+ * and must return a value (preferably unique) that will be item index.
+ */
+final class IndexWithReader implements
+    ItemReaderInterface,
+    InitializableInterface,
+    FlushableInterface,
+    JobExecutionAwareInterface
+{
+    use ElementConfiguratorTrait;
+    use JobExecutionAwareTrait;
+
+    private ItemReaderInterface $reader;
+    private Closure $extractItemIndex;
+
+    public function __construct(ItemReaderInterface $reader, Closure $extractItemIndex)
+    {
+        $this->reader = $reader;
+        $this->extractItemIndex = $extractItemIndex;
+    }
+
+    /**
+     * Uses item array value as the item index.
+     *
+     * Example, IndexWithReader::withArrayKey(..., 'name')
+     * will use 'name' array index of each read item as the item index.
+     */
+    public static function withArrayKey(ItemReaderInterface $reader, string $key): self
+    {
+        return new self($reader, fn(array $item) => $item[$key]);
+    }
+
+    /**
+     * Uses object property value as the item index.
+     *
+     * Example, IndexWithReader::withProperty(..., 'name')
+     * will use 'name' object property of each read item as the item index.
+     */
+    public static function withProperty(ItemReaderInterface $reader, string $property): self
+    {
+        return new self($reader, fn(object $item) => $item->$property);
+    }
+
+    /**
+     * Uses object method return value as the item index.
+     *
+     * Example, IndexWithReader::withProperty(..., 'getName')
+     * will call 'getName()' method of each read item and uses the result as the item index.
+     */
+    public static function withGetter(ItemReaderInterface $reader, string $getter): self
+    {
+        return new self($reader, fn(object $item) => $item->$getter());
+    }
+
+    /**
+     * @inheritdoc
+     */
+    public function initialize(): void
+    {
+        $this->configureElementJobContext($this->reader, $this->jobExecution);
+        $this->initializeElement($this->reader);
+    }
+
+    /**
+     * @inheritdoc
+     */
+    public function read(): iterable
+    {
+        foreach ($this->reader->read() as $item) {
+            yield ($this->extractItemIndex)($item) => $item;
+        }
+    }
+
+    /**
+     * @inheritdoc
+     */
+    public function flush(): void
+    {
+        $this->flushElement($this->reader);
+    }
+}

src/batch/tests/Job/Item/ItemJobTest.php

@@ -86,6 +86,14 @@ public function testExecute(): void
         self::assertSame(3, $jobExecution->getSummary()->get('invalid'), '3 items were invalid');
         self::assertSame(9, $jobExecution->getSummary()->get('write'), '9 items were write');
 
+        $warnings = $jobExecution->getWarnings();
+        self::assertCount(3, $warnings);
+        foreach ([[0, 9, 10], [1, 10, 11], [2, 11, 12]] as [$warningIdx, $itemIdx, $paramValue]) {
+            self::assertSame('Item is greater than 9 got {value}', $warnings[$warningIdx]->getMessage());
+            self::assertSame(['{value}' => $paramValue], $warnings[$warningIdx]->getParameters());
+            self::assertSame(['itemIndex' => $itemIdx], $warnings[$warningIdx]->getContext());
+        }
+
         $expectedLogs = <<<LOGS
 reader::setJobExecution
 reader::setJobParameters

src/batch/tests/Job/Item/Reader/IndexWithReaderTest.php

@@ -0,0 +1,76 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Yokai\Batch\Tests\Job\Item\Reader;
+
+use ArrayIterator;
+use Generator;
+use PHPUnit\Framework\TestCase;
+use Yokai\Batch\Job\Item\Reader\IndexWithReader;
+use Yokai\Batch\Job\Item\Reader\StaticIterableReader;
+use Yokai\Batch\JobExecution;
+
+class IndexWithReaderTest extends TestCase
+{
+    /**
+     * @dataProvider provider
+     */
+    public function test(callable $factory, array $expected): void
+    {
+        /** @var IndexWithReader $reader */
+        $reader = $factory();
+        $reader->setJobExecution(JobExecution::createRoot('123456', 'testing'));
+        $reader->initialize();
+
+        $actual = [];
+        foreach ($reader->read() as $index => $item) {
+            $actual[$index] = $item;
+        }
+
+        $reader->flush();
+
+        self::assertSame($expected, $actual);
+    }
+
+    public function provider(): Generator
+    {
+        $john = ['name' => 'John', 'location' => 'Washington'];
+        $marie = ['name' => 'Marie', 'location' => 'London'];
+        yield 'Index with array key' => [
+            fn() => IndexWithReader::withArrayKey(
+                new StaticIterableReader([$john, $marie]),
+                'name'
+            ),
+            ['John' => $john, 'Marie' => $marie],
+        ];
+
+        $john = (object)$john;
+        $marie = (object)$marie;
+        yield 'Index with object property' => [
+            fn() => IndexWithReader::withProperty(
+                new StaticIterableReader([$john, $marie]),
+                'name'
+            ),
+            ['John' => $john, 'Marie' => $marie],
+        ];
+
+        $three = new ArrayIterator([1, 2, 3]);
+        $six = new ArrayIterator([1, 2, 3, 4, 5, 6]);
+        yield 'Index with object method' => [
+            fn() => IndexWithReader::withGetter(
+                new StaticIterableReader([$three, $six]),
+                'count'
+            ),
+            [3 => $three, 6 => $six],
+        ];
+
+        yield 'Index with arbitrary closure' => [
+            fn() => new IndexWithReader(
+                new StaticIterableReader([1, 2, 3]),
+                fn(int $value) => $value * $value
+            ),
+            [1 => 1, 4 => 2, 9 => 3],
+        ];
+    }
+}