Add the ability to process each Bulk Batch with a processor class

Author: frankkessler

composer.json

@@ -35,7 +35,8 @@
   },
   "autoload-dev": {
     "files": [
-      "tests/GuzzleServer.php"
+      "tests/GuzzleServer.php",
+      "tests/BulkBatchProcessor.php"
     ]
   }
 }

readme.md

@@ -377,6 +377,22 @@ if ($result->id) {
 }
 ```
 
+### Query with Class to Process each Batch
+
+```php
+$operationType = 'query';
+$objectType = 'Account';
+$objectData = 'SELECT Id, Name FROM Account LIMIT 10';
+
+$result = Salesforce::bulk()->runBatch($operationType, $objectType, $objectData,[
+    'batchProcessor' => CustomBulkBatchProcessor::class, //Class must implement Frankkessler\Salesforce\Interfaces\BulkBatchProcessorInterface
+]);
+
+if ($result->id) {
+    $id = $result->id;
+}
+```
+
 ### Custom REST Endpoint (GET)
 
 ```php

src/Bulk.php

@@ -4,6 +4,7 @@
 
 use Frankkessler\Salesforce\Client\BulkClient;
 use Frankkessler\Salesforce\DataObjects\BinaryBatch;
+use Frankkessler\Salesforce\Interfaces\BulkBatchProcessorInterface;
 use Frankkessler\Salesforce\Responses\Bulk\BulkBatchResponse;
 use Frankkessler\Salesforce\Responses\Bulk\BulkBatchResultResponse;
 use Frankkessler\Salesforce\Responses\Bulk\BulkJobResponse;
@@ -45,6 +46,7 @@ public function runBatch($operation, $objectType, $data, $options = [])
             'isBatchedResult'           => false,
             'concurrencyMode'           => 'Parallel',
             'Sforce-Enable-PKChunking'  => false,
+            'batchProcessor'    => null,
         ];
 
         $options = array_replace($defaults, $options);
@@ -91,7 +93,11 @@ public function runBatch($operation, $objectType, $data, $options = [])
                 if (in_array($batch->state, ['Completed', 'Failed', 'Not Processed', 'NotProcessed'])) {
                     if (in_array($batch->state, ['Completed'])) {
                         $batchResult = $this->batchResult($job->id, $batch->id, $options['isBatchedResult'], null, $options['contentType']);
-                        $batch->records = $batchResult->records;
+                        if($options['batchProcessor'] instanceof BulkBatchProcessorInterface){
+                            call_user_func([$options['batchProcessor'], 'process'], $batchResult);
+                        }else {
+                            $batch->records = $batchResult->records;
+                        }
                     }
                     $batches_finished[] = $batch->id;
                 }

src/Interfaces/BulkBatchProcessorInterface.php

@@ -0,0 +1,10 @@
+<?php
+
+namespace Frankkessler\Salesforce\Interfaces;
+
+use Frankkessler\Salesforce\Responses\Bulk\BulkBatchResultResponse;
+
+interface BulkBatchProcessorInterface
+{
+    public static function process(BulkBatchResultResponse $batchResult);
+}

tests/BulkBatchProcessor.php

@@ -0,0 +1,14 @@
+<?php
+
+use Frankkessler\Salesforce\Interfaces\BulkBatchProcessorInterface;
+use Frankkessler\Salesforce\Responses\Bulk\BulkBatchResultResponse;
+
+class BulkBatchProcessor implements BulkBatchProcessorInterface
+{
+    public static $records = [];
+
+    public static function process(BulkBatchResultResponse $batchResult)
+    {
+        array_merge(static::$records, $batchResult->records);
+    }
+}

tests/BulkTest.php

@@ -255,6 +255,50 @@ public function testRunPkChunkQueryBatch()
         }
     }
 
+    public function testRunPkChunkQueryBatchWithBatchProcessor()
+    {
+        // Create a mock and queue two responses.
+        $mock = new MockHandler([
+            new Response(200, [], json_encode($this->jobArray())),
+            new Response(200, [], $this->batchXml()),
+            new Response(200, [], $this->batchListXml()),
+            new Response(200, [], $this->batchXml('NotProcessed')),
+            new Response(200, [], $this->batchXml('Completed', '8914000000B5YcIAAV', '89040000006iLsVAAU')),
+            new Response(200, [], $this->dataQueryResultXml()),
+            new Response(200, [], $this->dataQueryDataResultCsv()),
+            new Response(200, [], json_encode($this->jobArray())),
+        ]);
+
+        $handler = HandlerStack::create($mock);
+
+        $salesforce = new \Frankkessler\Salesforce\Salesforce([
+            'handler'                        => $handler,
+            'salesforce.oauth.access_token'  => 'TEST',
+            'salesforce.oauth.refresh_token' => 'TEST',
+        ]);
+
+        $jobId = '750D00000004SkVIAU';
+        $batchId = '8914000000B5YcIAAV';
+        $firstAccountId = '0014000001iM8r3AAC';
+
+        $operation = 'query';
+        $objectType = 'Account';
+        $query = "SELECT Id, Name FROM Account WHERE RecordTypeId='xxxxxxxxxxxxxx'";
+
+        $job = $salesforce->bulk()->runBatch($operation, $objectType, $query, [
+            'contentType'              => 'CSV',
+            'Sforce-Enable-PKChunking' => [
+                'chunkSize' => 2500,
+            ],
+            'batchProcessor'           => BulkBatchProcessor::class,
+        ]);
+
+        foreach(BulkBatchProcessor::$records as $record){
+            $this->assertEquals($firstAccountId, $record['Id']);
+            break;
+        }
+    }
+
     public function testBulkBinaryJobCreateWithMiddleware()
     {
         copy(realpath('./tests/bulk_zip_files').'/zipped_dir.zip', realpath('./build').'/test.zip');