wp-stateless / lib / classes / batch / class-batch-task-manager.php

class-batch-task-manager.php in WP-Stateless – Google Cloud Storage trunk, at lib/classes/batch/class-batch-task-manager.php

363 lines 9.0 KB

Raw Download Zip

1	<?php
2	/**
3	* Batch task manager
4	*
5	* @since 4.0.0
6	*/
7
8	namespace wpCloud\StatelessMedia\Batch;
9
10	// Require lib classes if not yet available
11	if (!class_exists('UDX_WP_Async_Request')) {
12	require_once ud_get_stateless_media()->path('lib/ns-vendor/classes/deliciousbrains/wp-background-processing/classes/wp-async-request.php', 'dir');
13	}
14
15	if (!class_exists('UDX_WP_Background_Process')) {
16	require_once ud_get_stateless_media()->path('lib/ns-vendor/classes/deliciousbrains/wp-background-processing/classes/wp-background-process.php', 'dir');
17	}
18
19	use wpCloud\StatelessMedia\Helper;
20	use wpCloud\StatelessMedia\Singleton;
21
22	class BatchTaskManager extends \UDX_WP_Background_Process {
23	use Singleton;
24
25	const STATE_KEY = '_state';
26	const UPDATED_KEY = '_updated';
27	const HEALTH_CHECK_INTERVAL = 60 * 5; // 5 minute
28
29	protected $prefix = 'sm';
30	protected $action = 'batch_process';
31
32	protected function __construct() {
33	parent::__construct();
34
35	$this->_init_hooks();
36	$this->_check_force_continue();
37	}
38
39	private function _init_hooks() {
40	add_filter('wp_stateless_batch_state', [$this, 'get_state'], 10, 1);
41	add_filter('wp_stateless_batch_action_pause', [$this, 'pause_task'], 10, 2);
42	add_filter('wp_stateless_batch_action_resume', [$this, 'resume_task'], 10, 2);
43	add_filter('heartbeat_send', [$this, 'check_running_batch'], 10, 1 );
44	}
45
46	/**
47	* Check if we should force dispatch
48	* Check if the task is in progress and if the state was not updated last 5 minutes - try to continue the task
49	*/
50	private function _check_force_continue() {
51	$last_updated = $this->_get_last_updated();
52
53	if ( empty($last_updated) \|\| $this->is_paused() ) {
54	return;
55	}
56
57	$check_interval = self::HEALTH_CHECK_INTERVAL;
58
59	if ( defined('WP_STATELESS_BATCH_HEALTHCHECK_INTERVAL') ) {
60	$check_interval = max($check_interval, WP_STATELESS_BATCH_HEALTHCHECK_INTERVAL * 60);
61	}
62
63	if ( time() - $last_updated <= $check_interval ) {
64	return;
65	}
66
67	Helper::log('Batch task freezed, trying to continue...');
68
69	// Forcing continue
70	$this->unlock_process();
71	$this->handle();
72	}
73
74	/**
75	* Update current task state
76	*
77	* @return array
78	*/
79	private function _update_state($state) {
80	update_option( $this->identifier . self::STATE_KEY, $state );
81	update_option( $this->identifier . self::UPDATED_KEY, time() );
82	}
83
84	/**
85	* Get current task state
86	*
87	* @return array
88	*/
89	private function _get_state() {
90	// We need to omit the cache and get the data directly from the db
91	global $wpdb;
92
93	$sql = "SELECT option_value FROM $wpdb->options WHERE option_name = '%s' LIMIT 1";
94	$sql = $wpdb->prepare($sql, $this->identifier . self::STATE_KEY);
95	$state = $wpdb->get_var($sql);
96
97	return empty($state) ? [] : maybe_unserialize($state);
98	}
99
100	/**
101	* Get last state update of the current task
102	*
103	* @return int\|null
104	*/
105	private function _get_last_updated() {
106	// We need to omit the cache and get the data directly from the db
107	global $wpdb;
108
109	$sql = "SELECT option_value FROM $wpdb->options WHERE option_name = '%s' LIMIT 1";
110	$sql = $wpdb->prepare($sql, $this->identifier . self::UPDATED_KEY);
111
112	return $wpdb->get_var($sql);
113	}
114
115	/**
116	* Delete current task state
117	*
118	* @return array
119	*/
120	private function _delete_state() {
121	delete_option( $this->identifier . self::STATE_KEY );
122	delete_option( $this->identifier . self::UPDATED_KEY );
123	}
124
125	/**
126	* Add new batch to the queue
127	*
128	* @param array $batch
129	*/
130	private function _add_batch($batch) {
131	if ( !empty($batch) ) {
132	$this->data( $batch )->save();
133	}
134	}
135
136	/**
137	* Get task object
138	*
139	* @param string $state\|null
140	* @return IBatchTask
141	* @throws \Exception
142	*/
143	private function _get_batch_task_object($state = null) {
144	if ( empty($state) ) {
145	$state = $this->_get_state();
146	}
147
148	if ( !isset($state['class']) \|\| !isset($state['file']) ) {
149	throw new \Exception("Can not get batch task file and class");
150	}
151
152	$class = $state['class'];
153
154	if ( !class_exists($class) ) {
155	require_once $state['file'];
156	}
157
158	$object = new $class();
159
160	if ( !is_a($object, '\wpCloud\StatelessMedia\Batch\IBatchTask') ) {
161	throw new \Exception("Batch task $class is not valid");
162	}
163
164	$object->set_state($state);
165
166	return $object;
167	}
168
169	/**
170	* Start the batch task
171	*
172	* @param string $class
173	* @param string\|null $file
174	* @param string $email
175	*/
176	public function start_task($class, $file = null, $email = '', $queue = []) {
177	try {
178	// Prepare default state
179	$defaults = [
180	'class' => $class,
181	'file' => $file,
182	'email' => $email,
183	'queue' => $queue,
184	];
185
186	$task_object = $this->_get_batch_task_object($defaults);
187	$task_object->init_state();
188
189	// Batch should be run prior to 'get_state' because it mutates the state
190	$this->_add_batch( $task_object->get_batch() );
191
192	// Save state
193	$state = wp_parse_args($task_object->get_state(), $defaults);
194
195	$this->_update_state( $state );
196
197	Helper::log('Batch task started: ' . $class);
198
199	do_action('wp_stateless_batch_task_started', $class, $file);
200	} catch (\Throwable $e) {
201	Helper::log("Batch task $class failed to start: " . $e->getMessage());
202
203	do_action('wp_stateless_batch_task_failed', $class, $file, $e->getMessage());
204
205	return;
206	}
207
208	$this->dispatch();
209	}
210
211	/**
212	* Process batch task item.
213	* Returns false to remove item from queue
214	* Returns $item to repeat
215	*
216	* @param string $item
217	* @return bool\|mixed
218	*/
219	public function task($item) {
220	$result = false;
221
222	try {
223	$object = $this->_get_batch_task_object();
224
225	$result = $object->process_item($item);
226	$this->_update_state( $object->get_state() );
227
228	$result = apply_filters('wp_stateless_batch_task_item_processed', $result, $item);
229	} catch (\Throwable $e) {
230	Helper::log( "Batch task unable to handle item $item: " . $e->getMessage() );
231
232	$result = apply_filters('wp_stateless_batch_task_item_failed', $result, $item);
233	}
234
235	return $result;
236	}
237
238	/**
239	* Complete the batch task. Tries to get the next batch and continue
240	*/
241	protected function complete() {
242	$class = '';
243
244	// Check if we have more batched to run
245	try {
246	$object = $this->_get_batch_task_object();
247	$class = get_class($object);
248	$batch = $object->get_batch();
249
250	if ( !empty($batch) ) {
251	$this->_add_batch( $batch );
252	$this->_update_state( $object->get_state() );
253
254	$this->dispatch();
255
256	return;
257	}
258
259	Helper::log( 'Batch task completed: ' . $class );
260	} catch (\Throwable $e) {
261	Helper::log( "Unable to process next batch: " . $e->getMessage() );
262	}
263
264	// If no more batches - delete state
265	$state = $this->_get_state();
266
267	parent::complete();
268	$this->_delete_state();
269
270	do_action('wp_stateless_batch_task_finished', $class, $state);
271
272	$site = site_url();
273	$subject = sprintf( __('WP-Stateless: Data Optimization Complete', ud_get_stateless_media()->domain) );
274	$message = sprintf(
275	__("WP-Stateless data has been optimized for %s.\n\nIf you have WP_STATELESS_SYNC_LOG or WP_DEBUG_LOG enabled, review those logs now to review any errors that may have occurred during the synchronization process.", ud_get_stateless_media()->domain),
276	$site
277	);
278
279	do_action('wp_stateless_send_admin_email', $subject, $message, $state['email'] ?? '');
280	}
281
282	/**
283	* Check if batch task has a state, so it is in progress
284	* Because is_processing is true only while processing an item
285	*
286	* @param array\|null $state
287	* @return bool
288	*/
289	public function is_running($state = null) {
290	if ( empty($state) ) {
291	$state = $this->_get_state();
292	}
293
294	return !empty($state);
295	}
296
297	/**
298	* Get the state of the current batch process
299	*
300	* @param mixed $status
301	* @return mixed
302	*/
303	public function get_state($state) {
304	$state = $this->_get_state();
305
306	unset($state['class']);
307	unset($state['file']);
308
309	$state['is_running'] = $this->is_running($state);
310	$state['is_paused'] = $this->is_paused();
311
312	return $state;
313	}
314
315	/**
316	* Pause the batch task
317	*
318	* @param array $state
319	* @param array $params
320	* @return array
321	*/
322	public function pause_task($state, $params) {
323	$this->pause();
324
325	return apply_filters('wp_stateless_batch_state', $state, []);
326	}
327
328	/**
329	* Resume the batch task
330	*
331	* @param array $state
332	* @param array $params
333	* @return array
334	*/
335	public function resume_task($state, $params) {
336	$this->resume();
337
338	return apply_filters('wp_stateless_batch_state', $state, []);
339	}
340
341	/**
342	* Get the state key
343	*
344	* @return string
345	*/
346	public function get_state_key() {
347	return $this->identifier . self::STATE_KEY;
348	}
349
350	/**
351	* Check if batch is running during WP heartbeat request
352	*
353	* @return array
354	*/
355	public function check_running_batch($response) {
356	if ( $this->is_running() ) {
357	$response['stateless-batch-running'] = true;
358	}
359
360	return $response;
361	}
362	}
363