PHP Classes
elePHPant
Icontem

File: application/modules/extensions/aws/Aws/DynamoDb/WriteRequestBatch.php

Recommend this page to a friend!
  Classes of Tran Tuan  >  Pretty PHP S3 Files Manager  >  application/modules/extensions/aws/Aws/DynamoDb/WriteRequestBatch.php  >  Download  
File: application/modules/extensions/aws/Aws/DynamoDb/WriteRequestBatch.php
Role: Application script
Content type: text/plain
Description: Application script
Class: Pretty PHP S3 Files Manager
Web based interface to manage files in Amazon S3
Author: By
Last change:
Date: 3 years ago
Size: 9,591 bytes
 

 

Contents

Class file image Download
<?php
namespace Aws\DynamoDb;

use
Aws\AwsClientInterface;
use
Aws\CommandInterface;
use
Aws\CommandPool;
use
Aws\Exception\AwsException;
use
Aws\ResultInterface;

/**
 * The WriteRequestBatch is an object that is capable of efficiently sending
 * DynamoDB BatchWriteItem requests from queued up put and delete item requests.
 * requests. The batch attempts to send the requests with the fewest requests
 * to DynamoDB as possible and also re-queues any unprocessed items to ensure
 * that all items are sent.
 */
class WriteRequestBatch
{
   
/** @var DynamoDbClient DynamoDB client used to perform write operations. */
   
private $client;

   
/** @var array Configuration options for the batch. */
   
private $config;

   
/** @var array Queue of pending put/delete requests in the batch. */
   
private $queue;

   
/**
     * Creates a WriteRequestBatch object that is capable of efficiently sending
     * DynamoDB BatchWriteItem requests from queued up Put and Delete requests.
     *
     * @param DynamoDbClient $client DynamoDB client used to send batches.
     * @param array $config Batch configuration options.
     * - table: (string) DynamoDB table used by the batch, this can be
     * overridden for each individual put() or delete() call.
     * - batch_size: (int) The size of each batch (default: 25). The batch
     * size must be between 2 and 25. If you are sending batches of large
     * items, you may consider lowering the batch size, otherwise, you
     * should use 25.
     * - pool_size: (int) This number dictates how many BatchWriteItem
     * requests you would like to do in parallel. For example, if the
     * "batch_size" is 25, and "pool_size" is 3, then you would send 3
     * BatchWriteItem requests at a time, each with 25 items. Please keep
     * your throughput in mind when choosing the "pool_size" option.
     * - autoflush: (bool) This option allows the batch to automatically
     * flush once there are enough items (i.e., "batch_size" * "pool_size")
     * in the queue. This defaults to true, so you must set this to false
     * to stop autoflush.
     * - before: (callable) Executed before every BatchWriteItem operation.
     * It should accept an \Aws\CommandInterface object as its argument.
     * - error: Executed if an error was encountered executing a,
     * BatchWriteItem operation, otherwise errors are ignored. It should
     * accept an \Aws\Exception\AwsException as its argument.
     *
     * @throws \InvalidArgumentException if the batch size is not between 2 and 25.
     */
   
public function __construct(DynamoDbClient $client, array $config = [])
    {
       
// Apply defaults
       
$config += [
           
'table' => null,
           
'batch_size' => 25,
           
'pool_size' => 1,
           
'autoflush' => true,
           
'before' => null,
           
'error' => null
       
];

       
// Ensure the batch size is valid
       
if ($config['batch_size'] > 25 || $config['batch_size'] < 2) {
            throw new \
InvalidArgumentException('"batch_size" must be between 2 and 25.');
        }

       
// Ensure the callbacks are valid
       
if ($config['before'] && !is_callable($config['before'])) {
            throw new \
InvalidArgumentException('"before" must be callable.');
        }
        if (
$config['error'] && !is_callable($config['error'])) {
            throw new \
InvalidArgumentException('"error" must be callable.');
        }

       
// If autoflush is enabled, set the threshold
       
if ($config['autoflush']) {
           
$config['threshold'] = $config['batch_size'] * $config['pool_size'];
        }

       
$this->client = $client;
       
$this->config = $config;
       
$this->queue = [];
    }

   
/**
     * Adds a put item request to the batch.
     *
     * @param array $item Data for an item to put. Format:
     * [
     * 'attribute1' => ['type' => 'value'],
     * 'attribute2' => ['type' => 'value'],
     * ...
     * ]
     * @param string|null $table The name of the table. This must be specified
     * unless the "table" option was provided in the
     * config of the WriteRequestBatch.
     *
     * @return $this
     */
   
public function put(array $item, $table = null)
    {
       
$this->queue[] = [
           
'table' => $this->determineTable($table),
           
'data' => ['PutRequest' => ['Item' => $item]],
        ];

       
$this->autoFlush();

        return
$this;
    }

   
/**
     * Adds a delete item request to the batch.
     *
     * @param array $key Key of an item to delete. Format:
     * [
     * 'key1' => ['type' => 'value'],
     * ...
     * ]
     * @param string|null $table The name of the table. This must be specified
     * unless the "table" option was provided in the
     * config of the WriteRequestBatch.
     *
     * @return $this
     */
   
public function delete(array $key, $table = null)
    {
       
$this->queue[] = [
           
'table' => $this->determineTable($table),
           
'data' => ['DeleteRequest' => ['Key' => $key]],
        ];

       
$this->autoFlush();

        return
$this;
    }

   
/**
     * Flushes the batch by combining all the queued put and delete requests
     * into BatchWriteItem commands and executing them. Unprocessed items are
     * automatically re-queued.
     *
     * @param bool $untilEmpty If true, flushing will continue until the queue
     * is completely empty. This will make sure that
     * unprocessed items are all eventually sent.
     *
     * @return $this
     */
   
public function flush($untilEmpty = true)
    {
       
// Send BatchWriteItem requests until the queue is empty
       
$keepFlushing = true;
        while (
$this->queue && $keepFlushing) {
           
$commands = $this->prepareCommands();
           
$pool = new CommandPool($this->client, $commands, [
               
'before' => $this->config['before'],
               
'concurrency' => $this->config['pool_size'],
               
'fulfilled' => function (ResultInterface $result) {
                   
// Re-queue any unprocessed items
                   
if ($result->hasKey('UnprocessedItems')) {
                       
$this->retryUnprocessed($result['UnprocessedItems']);
                    }
                },
               
'rejected' => function ($reason) {
                    if (
$reason instanceof AwsException) {
                       
$code = $reason->getAwsErrorCode();
                        if (
$code === 'ProvisionedThroughputExceededException') {
                           
$this->retryUnprocessed($reason->getCommand()['RequestItems']);
                        } elseif (
is_callable($this->config['error'])) {
                           
$this->config['error']($reason);
                        }
                    }
                }
            ]);
           
$pool->promise()->wait();
           
$keepFlushing = (bool) $untilEmpty;
        }

        return
$this;
    }

   
/**
     * Creates BatchWriteItem commands from the items in the queue.
     *
     * @return CommandInterface[]
     */
   
private function prepareCommands()
    {
       
// Chunk the queue into batches
       
$batches = array_chunk($this->queue, $this->config['batch_size']);
       
$this->queue = [];

       
// Create BatchWriteItem commands for each batch
       
$commands = [];
        foreach (
$batches as $batch) {
           
$requests = [];
            foreach (
$batch as $item) {
                if (!isset(
$requests[$item['table']])) {
                   
$requests[$item['table']] = [];
                }
               
$requests[$item['table']][] = $item['data'];
            }
           
$commands[] = $this->client->getCommand(
               
'BatchWriteItem',
                [
'RequestItems' => $requests]
            );
        }

        return
$commands;
    }

   
/**
     * Re-queues unprocessed results with the correct data.
     *
     * @param array $unprocessed Unprocessed items from a result.
     */
   
private function retryUnprocessed(array $unprocessed)
    {
        foreach (
$unprocessed as $table => $requests) {
            foreach (
$requests as $request) {
               
$this->queue[] = [
                   
'table' => $table,
                   
'data' => $request,
                ];
            }
        }
    }

   
/**
     * If autoflush is enabled and the threshold is met, flush the batch
     */
   
private function autoFlush()
    {
        if (
$this->config['autoflush']
            &&
count($this->queue) >= $this->config['threshold']
        ) {
           
// Flush only once. Unprocessed items are handled in a later flush.
           
$this->flush(false);
        }
    }

   
/**
     * Determine the table name by looking at what was provided and what the
     * WriteRequestBatch was originally configured with.
     *
     * @param string|null $table The table name.
     *
     * @return string
     * @throws \RuntimeException if there was no table specified.
     */
   
private function determineTable($table)
    {
       
$table = $table ?: $this->config['table'];
        if (!
$table) {
            throw new \
RuntimeException('There was no table specified.');
        }

        return
$table;
    }
}