jetpack / jetpack_vendor / automattic / jetpack-sync / src / class-rest-sender.php

class-rest-sender.php in Jetpack – WP Security, Backup, Speed, & Growth 16.3-a.7, at jetpack_vendor/automattic/jetpack-sync/src/class-rest-sender.php

153 lines 4.4 KB

Raw Download Zip

1	<?php
2	/**
3	* Sync package.
4	*
5	* @package automattic/jetpack-sync
6	*/
7
8	namespace Automattic\Jetpack\Sync;
9
10	use WP_Error;
11
12	/**
13	* This class will handle checkout of Sync queues for REST Endpoints.
14	*
15	* @since 1.23.1
16	*/
17	class REST_Sender {
18
19	/**
20	* Items pending send.
21	*
22	* @var array
23	*/
24	public $items = array();
25
26	/**
27	* Checkout objects from the queue
28	*
29	* @param string $queue_name Name of Queue.
30	* @param int\|null $number_of_items Number of Items. Null for memory-based checkout.
31	* @param array $args Request arguments. Supports 'pop', 'force', 'encode', and 'use_memory_limit'.
32	*
33	* @return array\|WP_Error
34	*/
35	public function queue_pull( $queue_name, $number_of_items, $args ) {
36	$queue = new Queue( $queue_name );
37
38	if ( 0 === $queue->size() ) {
39	return new WP_Error( 'queue_size', 'The queue is empty and there is nothing to send', 400 );
40	}
41
42	$sender = Sender::get_instance();
43
44	// try to give ourselves as much time as possible.
45	set_time_limit( 0 );
46
47	if ( ! empty( $args['pop'] ) ) {
48	$buffer = new Queue_Buffer( 'pop', $queue->pop( $number_of_items ) );
49	} else {
50	// let's delete the checkin state.
51	if ( $args['force'] ) {
52	$queue->unlock();
53	}
54	$buffer = $this->get_buffer( $queue, $number_of_items );
55	}
56	// Check that the $buffer is not checkout out already.
57	if ( is_wp_error( $buffer ) ) {
58	return new WP_Error( 'buffer_open', "We couldn't get the buffer it is currently checked out", 400 );
59	}
60
61	if ( ! is_object( $buffer ) ) {
62	return new WP_Error( 'buffer_non-object', 'Buffer is not an object', 400 );
63	}
64
65	$encode = $args['encode'] ?? true;
66
67	Settings::set_is_syncing( true );
68	list( $items_to_send, $skipped_items_ids ) = $sender->get_items_to_send( $buffer, $encode );
69	Settings::set_is_syncing( false );
70
71	return array(
72	'buffer_id' => $buffer->id,
73	'items' => $items_to_send,
74	'skipped_items' => $skipped_items_ids,
75	'codec' => $encode ? $sender->get_codec()->name() : null,
76	'sent_timestamp' => time(),
77	'queue_size' => $queue->size(),
78	);
79	}
80
81	/**
82	* Adds Sync items to local property.
83	*/
84	public function jetpack_sync_send_data_listener() {
85	foreach ( func_get_args()[0] as $key => $item ) {
86	$this->items[ $key ] = $item;
87	}
88	}
89
90	/**
91	* Check out a buffer of full sync actions.
92	*
93	* @return array Sync Actions to be returned to requestor
94	*/
95	public function immediate_full_sync_pull() {
96	// try to give ourselves as much time as possible.
97	set_time_limit( 0 );
98
99	$original_send_data_cb = array( 'Automattic\Jetpack\Sync\Actions', 'send_data' );
100	$temp_send_data_cb = array( $this, 'jetpack_sync_send_data_listener' );
101
102	Sender::get_instance()->set_enqueue_wait_time( 0 );
103	remove_filter( 'jetpack_sync_send_data', $original_send_data_cb );
104	add_filter( 'jetpack_sync_send_data', $temp_send_data_cb, 10, 6 );
105	Sender::get_instance()->do_full_sync();
106	remove_filter( 'jetpack_sync_send_data', $temp_send_data_cb );
107	add_filter( 'jetpack_sync_send_data', $original_send_data_cb, 10, 6 );
108
109	return array(
110	'items' => $this->items,
111	'codec' => Sender::get_instance()->get_codec()->name(),
112	'sent_timestamp' => time(),
113	'status' => Actions::get_sync_status(),
114	);
115	}
116
117	/**
118	* Checkout items out of the sync queue.
119	*
120	* @param Queue $queue Sync Queue.
121	* @param int\|null $number_of_items Number of items to checkout. When null, uses memory-based checkout with default settings.
122	*
123	* @return Queue_Buffer\|WP_Error
124	*/
125	protected function get_buffer( $queue, $number_of_items ) {
126	$start = time();
127	$max_duration = 5; // this will try to get the buffer.
128
129	$use_memory_limit = null === $number_of_items;
130	$memory_limit = $use_memory_limit ? Settings::get_setting( 'dequeue_max_bytes' ) : null;
131	$max_rows = $use_memory_limit ? Settings::get_setting( 'upload_max_rows' ) : null;
132
133	$buffer = $use_memory_limit
134	? $queue->checkout_with_memory_limit( $memory_limit, $max_rows )
135	: $queue->checkout( $number_of_items );
136	$duration = time() - $start;
137
138	while ( is_wp_error( $buffer ) && $duration < $max_duration ) {
139	sleep( 2 );
140	$duration = time() - $start;
141	$buffer = $use_memory_limit
142	? $queue->checkout_with_memory_limit( $memory_limit, $max_rows )
143	: $queue->checkout( $number_of_items );
144	}
145
146	if ( false === $buffer ) {
147	return new WP_Error( 'queue_size', 'The queue is empty and there is nothing to send', 400 );
148	}
149
150	return $buffer;
151	}
152	}
153