Exception
11 months ago
BinaryValue.php
11 months ago
DynamoDbClient.php
11 months ago
LockingSessionConnection.php
11 months ago
Marshaler.php
11 months ago
NumberValue.php
11 months ago
SessionConnectionConfigTrait.php
11 months ago
SessionConnectionInterface.php
11 months ago
SessionHandler.php
11 months ago
SetValue.php
11 months ago
StandardSessionConnection.php
11 months ago
WriteRequestBatch.php
11 months ago
WriteRequestBatch.php
267 lines
| 1 | <?php |
| 2 | namespace Aws\DynamoDb; |
| 3 | |
| 4 | use Aws\CommandInterface; |
| 5 | use Aws\CommandPool; |
| 6 | use Aws\Exception\AwsException; |
| 7 | use Aws\ResultInterface; |
| 8 | |
| 9 | /** |
| 10 | * The WriteRequestBatch is an object that is capable of efficiently sending |
| 11 | * DynamoDB BatchWriteItem requests from queued up put and delete item requests. |
| 12 | * requests. The batch attempts to send the requests with the fewest requests |
| 13 | * to DynamoDB as possible and also re-queues any unprocessed items to ensure |
| 14 | * that all items are sent. |
| 15 | */ |
| 16 | class WriteRequestBatch |
| 17 | { |
| 18 | /** @var DynamoDbClient DynamoDB client used to perform write operations. */ |
| 19 | private $client; |
| 20 | |
| 21 | /** @var array Configuration options for the batch. */ |
| 22 | private $config; |
| 23 | |
| 24 | /** @var array Queue of pending put/delete requests in the batch. */ |
| 25 | private $queue; |
| 26 | |
| 27 | /** |
| 28 | * Creates a WriteRequestBatch object that is capable of efficiently sending |
| 29 | * DynamoDB BatchWriteItem requests from queued up Put and Delete requests. |
| 30 | * |
| 31 | * @param DynamoDbClient $client DynamoDB client used to send batches. |
| 32 | * @param array $config Batch configuration options. |
| 33 | * - table: (string) DynamoDB table used by the batch, this can be |
| 34 | * overridden for each individual put() or delete() call. |
| 35 | * - batch_size: (int) The size of each batch (default: 25). The batch |
| 36 | * size must be between 2 and 25. If you are sending batches of large |
| 37 | * items, you may consider lowering the batch size, otherwise, you |
| 38 | * should use 25. |
| 39 | * - pool_size: (int) This number dictates how many BatchWriteItem |
| 40 | * requests you would like to do in parallel. For example, if the |
| 41 | * "batch_size" is 25, and "pool_size" is 3, then you would send 3 |
| 42 | * BatchWriteItem requests at a time, each with 25 items. Please keep |
| 43 | * your throughput in mind when choosing the "pool_size" option. |
| 44 | * - autoflush: (bool) This option allows the batch to automatically |
| 45 | * flush once there are enough items (i.e., "batch_size" * "pool_size") |
| 46 | * in the queue. This defaults to true, so you must set this to false |
| 47 | * to stop autoflush. |
| 48 | * - before: (callable) Executed before every BatchWriteItem operation. |
| 49 | * It should accept an \Aws\CommandInterface object as its argument. |
| 50 | * - error: Executed if an error was encountered executing a, |
| 51 | * BatchWriteItem operation, otherwise errors are ignored. It should |
| 52 | * accept an \Aws\Exception\AwsException as its argument. |
| 53 | * |
| 54 | * @throws \InvalidArgumentException if the batch size is not between 2 and 25. |
| 55 | */ |
| 56 | public function __construct(DynamoDbClient $client, array $config = []) |
| 57 | { |
| 58 | // Apply defaults |
| 59 | $config += [ |
| 60 | 'table' => null, |
| 61 | 'batch_size' => 25, |
| 62 | 'pool_size' => 1, |
| 63 | 'autoflush' => true, |
| 64 | 'before' => null, |
| 65 | 'error' => null |
| 66 | ]; |
| 67 | |
| 68 | // Ensure the batch size is valid |
| 69 | if ($config['batch_size'] > 25 || $config['batch_size'] < 2) { |
| 70 | throw new \InvalidArgumentException('"batch_size" must be between 2 and 25.'); |
| 71 | } |
| 72 | |
| 73 | // Ensure the callbacks are valid |
| 74 | if ($config['before'] && !is_callable($config['before'])) { |
| 75 | throw new \InvalidArgumentException('"before" must be callable.'); |
| 76 | } |
| 77 | if ($config['error'] && !is_callable($config['error'])) { |
| 78 | throw new \InvalidArgumentException('"error" must be callable.'); |
| 79 | } |
| 80 | |
| 81 | // If autoflush is enabled, set the threshold |
| 82 | if ($config['autoflush']) { |
| 83 | $config['threshold'] = $config['batch_size'] * $config['pool_size']; |
| 84 | } |
| 85 | |
| 86 | $this->client = $client; |
| 87 | $this->config = $config; |
| 88 | $this->queue = []; |
| 89 | } |
| 90 | |
| 91 | /** |
| 92 | * Adds a put item request to the batch. |
| 93 | * |
| 94 | * @param array $item Data for an item to put. Format: |
| 95 | * [ |
| 96 | * 'attribute1' => ['type' => 'value'], |
| 97 | * 'attribute2' => ['type' => 'value'], |
| 98 | * ... |
| 99 | * ] |
| 100 | * @param string|null $table The name of the table. This must be specified |
| 101 | * unless the "table" option was provided in the |
| 102 | * config of the WriteRequestBatch. |
| 103 | * |
| 104 | * @return $this |
| 105 | */ |
| 106 | public function put(array $item, $table = null) |
| 107 | { |
| 108 | $this->queue[] = [ |
| 109 | 'table' => $this->determineTable($table), |
| 110 | 'data' => ['PutRequest' => ['Item' => $item]], |
| 111 | ]; |
| 112 | |
| 113 | $this->autoFlush(); |
| 114 | |
| 115 | return $this; |
| 116 | } |
| 117 | |
| 118 | /** |
| 119 | * Adds a delete item request to the batch. |
| 120 | * |
| 121 | * @param array $key Key of an item to delete. Format: |
| 122 | * [ |
| 123 | * 'key1' => ['type' => 'value'], |
| 124 | * ... |
| 125 | * ] |
| 126 | * @param string|null $table The name of the table. This must be specified |
| 127 | * unless the "table" option was provided in the |
| 128 | * config of the WriteRequestBatch. |
| 129 | * |
| 130 | * @return $this |
| 131 | */ |
| 132 | public function delete(array $key, $table = null) |
| 133 | { |
| 134 | $this->queue[] = [ |
| 135 | 'table' => $this->determineTable($table), |
| 136 | 'data' => ['DeleteRequest' => ['Key' => $key]], |
| 137 | ]; |
| 138 | |
| 139 | $this->autoFlush(); |
| 140 | |
| 141 | return $this; |
| 142 | } |
| 143 | |
| 144 | /** |
| 145 | * Flushes the batch by combining all the queued put and delete requests |
| 146 | * into BatchWriteItem commands and executing them. Unprocessed items are |
| 147 | * automatically re-queued. |
| 148 | * |
| 149 | * @param bool $untilEmpty If true, flushing will continue until the queue |
| 150 | * is completely empty. This will make sure that |
| 151 | * unprocessed items are all eventually sent. |
| 152 | * |
| 153 | * @return $this |
| 154 | */ |
| 155 | public function flush($untilEmpty = true) |
| 156 | { |
| 157 | // Send BatchWriteItem requests until the queue is empty |
| 158 | $keepFlushing = true; |
| 159 | while ($this->queue && $keepFlushing) { |
| 160 | $commands = $this->prepareCommands(); |
| 161 | $pool = new CommandPool($this->client, $commands, [ |
| 162 | 'before' => $this->config['before'], |
| 163 | 'concurrency' => $this->config['pool_size'], |
| 164 | 'fulfilled' => function (ResultInterface $result) { |
| 165 | // Re-queue any unprocessed items |
| 166 | if ($result->hasKey('UnprocessedItems')) { |
| 167 | $this->retryUnprocessed($result['UnprocessedItems']); |
| 168 | } |
| 169 | }, |
| 170 | 'rejected' => function ($reason) { |
| 171 | if ($reason instanceof AwsException) { |
| 172 | $code = $reason->getAwsErrorCode(); |
| 173 | if ($code === 'ProvisionedThroughputExceededException') { |
| 174 | $this->retryUnprocessed($reason->getCommand()['RequestItems']); |
| 175 | } elseif (is_callable($this->config['error'])) { |
| 176 | $this->config['error']($reason); |
| 177 | } |
| 178 | } |
| 179 | } |
| 180 | ]); |
| 181 | $pool->promise()->wait(); |
| 182 | $keepFlushing = (bool) $untilEmpty; |
| 183 | } |
| 184 | |
| 185 | return $this; |
| 186 | } |
| 187 | |
| 188 | /** |
| 189 | * Creates BatchWriteItem commands from the items in the queue. |
| 190 | * |
| 191 | * @return CommandInterface[] |
| 192 | */ |
| 193 | private function prepareCommands() |
| 194 | { |
| 195 | // Chunk the queue into batches |
| 196 | $batches = array_chunk($this->queue, $this->config['batch_size']); |
| 197 | $this->queue = []; |
| 198 | |
| 199 | // Create BatchWriteItem commands for each batch |
| 200 | $commands = []; |
| 201 | foreach ($batches as $batch) { |
| 202 | $requests = []; |
| 203 | foreach ($batch as $item) { |
| 204 | if (!isset($requests[$item['table']])) { |
| 205 | $requests[$item['table']] = []; |
| 206 | } |
| 207 | $requests[$item['table']][] = $item['data']; |
| 208 | } |
| 209 | $commands[] = $this->client->getCommand( |
| 210 | 'BatchWriteItem', |
| 211 | ['RequestItems' => $requests] |
| 212 | ); |
| 213 | } |
| 214 | |
| 215 | return $commands; |
| 216 | } |
| 217 | |
| 218 | /** |
| 219 | * Re-queues unprocessed results with the correct data. |
| 220 | * |
| 221 | * @param array $unprocessed Unprocessed items from a result. |
| 222 | */ |
| 223 | private function retryUnprocessed(array $unprocessed) |
| 224 | { |
| 225 | foreach ($unprocessed as $table => $requests) { |
| 226 | foreach ($requests as $request) { |
| 227 | $this->queue[] = [ |
| 228 | 'table' => $table, |
| 229 | 'data' => $request, |
| 230 | ]; |
| 231 | } |
| 232 | } |
| 233 | } |
| 234 | |
| 235 | /** |
| 236 | * If autoflush is enabled and the threshold is met, flush the batch |
| 237 | */ |
| 238 | private function autoFlush() |
| 239 | { |
| 240 | if ($this->config['autoflush'] |
| 241 | && count($this->queue) >= $this->config['threshold'] |
| 242 | ) { |
| 243 | // Flush only once. Unprocessed items are handled in a later flush. |
| 244 | $this->flush(false); |
| 245 | } |
| 246 | } |
| 247 | |
| 248 | /** |
| 249 | * Determine the table name by looking at what was provided and what the |
| 250 | * WriteRequestBatch was originally configured with. |
| 251 | * |
| 252 | * @param string|null $table The table name. |
| 253 | * |
| 254 | * @return string |
| 255 | * @throws \RuntimeException if there was no table specified. |
| 256 | */ |
| 257 | private function determineTable($table) |
| 258 | { |
| 259 | $table = $table ?: $this->config['table']; |
| 260 | if (!$table) { |
| 261 | throw new \RuntimeException('There was no table specified.'); |
| 262 | } |
| 263 | |
| 264 | return $table; |
| 265 | } |
| 266 | } |
| 267 |