media-cloud-sync / includes / sdk / s3 / Aws / S3 / BatchDelete.php

BatchDelete.php in Media Cloud Sync 1.0.2, at includes/sdk/s3/Aws/S3/BatchDelete.php

197 lines 7.2 KB

Raw Download Zip

1	<?php
2
3	namespace Dudlewebs\WPMCS\s3\Aws\S3;
4
5	use Dudlewebs\WPMCS\s3\Aws\AwsClientInterface;
6	use Dudlewebs\WPMCS\s3\Aws\S3\Exception\DeleteMultipleObjectsException;
7	use Dudlewebs\WPMCS\s3\GuzzleHttp\Promise;
8	use Dudlewebs\WPMCS\s3\GuzzleHttp\Promise\PromisorInterface;
9	use Dudlewebs\WPMCS\s3\GuzzleHttp\Promise\PromiseInterface;
10	/**
11	* Efficiently deletes many objects from a single Amazon S3 bucket using an
12	* iterator that yields keys. Deletes are made using the DeleteObjects API
13	* operation.
14	*
15	* $s3 = new Aws\S3\Client([
16	* 'region' => 'us-west-2',
17	* 'version' => 'latest'
18	* ]);
19	*
20	* $listObjectsParams = ['Bucket' => 'foo', 'Prefix' => 'starts/with/'];
21	* $delete = Aws\S3\BatchDelete::fromListObjects($s3, $listObjectsParams);
22	* // Asynchronously delete
23	* $promise = $delete->promise();
24	* // Force synchronous completion
25	* $delete->delete();
26	*
27	* When using one of the batch delete creational static methods, you can supply
28	* an associative array of options:
29	*
30	* - before: Function invoked before executing a command. The function is
31	* passed the command that is about to be executed. This can be useful
32	* for logging, adding custom request headers, etc.
33	* - batch_size: The size of each delete batch. Defaults to 1000.
34	*
35	* @link http://docs.aws.amazon.com/AmazonS3/latest/API/multiobjectdeleteapi.html
36	*/
37	class BatchDelete implements PromisorInterface
38	{
39	private $bucket;
40	/** @var AwsClientInterface */
41	private $client;
42	/** @var callable */
43	private $before;
44	/** @var PromiseInterface */
45	private $cachedPromise;
46	/** @var callable */
47	private $promiseCreator;
48	private $batchSize = 1000;
49	private $queue = [];
50	/**
51	* Creates a BatchDelete object from all of the paginated results of a
52	* ListObjects operation. Each result that is returned by the ListObjects
53	* operation will be deleted.
54	*
55	* @param AwsClientInterface $client AWS Client to use.
56	* @param array $listObjectsParams ListObjects API parameters
57	* @param array $options BatchDelete options.
58	*
59	* @return BatchDelete
60	*/
61	public static function fromListObjects(AwsClientInterface $client, array $listObjectsParams, array $options = [])
62	{
63	$iter = $client->getPaginator('ListObjects', $listObjectsParams);
64	$bucket = $listObjectsParams['Bucket'];
65	$fn = function (BatchDelete $that) use($iter) {
66	return $iter->each(function ($result) use($that) {
67	$promises = [];
68	if (\is_array($result['Contents'])) {
69	foreach ($result['Contents'] as $object) {
70	if ($promise = $that->enqueue($object)) {
71	$promises[] = $promise;
72	}
73	}
74	}
75	return $promises ? Promise\Utils::all($promises) : null;
76	});
77	};
78	return new self($client, $bucket, $fn, $options);
79	}
80	/**
81	* Creates a BatchDelete object from an iterator that yields results.
82	*
83	* @param AwsClientInterface $client AWS Client to use to execute commands
84	* @param string $bucket Bucket where the objects are stored
85	* @param \Iterator $iter Iterator that yields assoc arrays
86	* @param array $options BatchDelete options
87	*
88	* @return BatchDelete
89	*/
90	public static function fromIterator(AwsClientInterface $client, $bucket, \Iterator $iter, array $options = [])
91	{
92	$fn = function (BatchDelete $that) use($iter) {
93	return Promise\Coroutine::of(function () use($that, $iter) {
94	foreach ($iter as $obj) {
95	if ($promise = $that->enqueue($obj)) {
96	(yield $promise);
97	}
98	}
99	});
100	};
101	return new self($client, $bucket, $fn, $options);
102	}
103	/**
104	* @return PromiseInterface
105	*/
106	public function promise()
107	{
108	if (!$this->cachedPromise) {
109	$this->cachedPromise = $this->createPromise();
110	}
111	return $this->cachedPromise;
112	}
113	/**
114	* Synchronously deletes all of the objects.
115	*
116	* @throws DeleteMultipleObjectsException on error.
117	*/
118	public function delete()
119	{
120	$this->promise()->wait();
121	}
122	/**
123	* @param AwsClientInterface $client Client used to transfer the requests
124	* @param string $bucket Bucket to delete from.
125	* @param callable $promiseFn Creates a promise.
126	* @param array $options Hash of options used with the batch
127	*
128	* @throws \InvalidArgumentException if the provided batch_size is <= 0
129	*/
130	private function __construct(AwsClientInterface $client, $bucket, callable $promiseFn, array $options = [])
131	{
132	$this->client = $client;
133	$this->bucket = $bucket;
134	$this->promiseCreator = $promiseFn;
135	if (isset($options['before'])) {
136	if (!\is_callable($options['before'])) {
137	throw new \InvalidArgumentException('before must be callable');
138	}
139	$this->before = $options['before'];
140	}
141	if (isset($options['batch_size'])) {
142	if ($options['batch_size'] <= 0) {
143	throw new \InvalidArgumentException('batch_size is not > 0');
144	}
145	$this->batchSize = \min($options['batch_size'], 1000);
146	}
147	}
148	private function enqueue(array $obj)
149	{
150	$this->queue[] = $obj;
151	return \count($this->queue) >= $this->batchSize ? $this->flushQueue() : null;
152	}
153	private function flushQueue()
154	{
155	static $validKeys = ['Key' => \true, 'VersionId' => \true];
156	if (\count($this->queue) === 0) {
157	return null;
158	}
159	$batch = [];
160	while ($obj = \array_shift($this->queue)) {
161	$batch[] = \array_intersect_key($obj, $validKeys);
162	}
163	$command = $this->client->getCommand('DeleteObjects', ['Bucket' => $this->bucket, 'Delete' => ['Objects' => $batch]]);
164	if ($this->before) {
165	\call_user_func($this->before, $command);
166	}
167	return $this->client->executeAsync($command)->then(function ($result) {
168	if (!empty($result['Errors'])) {
169	throw new DeleteMultipleObjectsException($result['Deleted'] ?: [], $result['Errors']);
170	}
171	return $result;
172	});
173	}
174	/**
175	* Returns a promise that will clean up any references when it completes.
176	*
177	* @return PromiseInterface
178	*/
179	private function createPromise()
180	{
181	// Create the promise
182	$promise = \call_user_func($this->promiseCreator, $this);
183	$this->promiseCreator = null;
184	// Cleans up the promise state and references.
185	$cleanup = function () {
186	$this->before = $this->client = $this->queue = null;
187	};
188	// When done, ensure cleanup and that any remaining are processed.
189	return $promise->then(function () use($cleanup) {
190	return Promise\Create::promiseFor($this->flushQueue())->then($cleanup);
191	}, function ($reason) use($cleanup) {
192	$cleanup();
193	return Promise\Create::rejectionFor($reason);
194	});
195	}
196	}
197