Abstract_Jetpack_Sync_Queue - Code Metrics - Inspection of "Revert unrelated change to sync queue" - Automattic/jetpack - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Push — fix/vvv-tests-wp-5-2 ( ec7a8b...e4b178 )

unknown

created 2019-05-17 21:53 UTC

Abstract_Jetpack_Sync_Queue A

↳ Parent: Project

Complexity

Total Complexity

Size/Duplication

Total Lines	6
Duplicated Lines	0 %

Coupling/Cohesion

Components	0
Dependencies	0

Importance

Changes

Metric	Value
dl	0
loc	6
rs	10
c	0
b	0
f	0
wmc	1
lcom	0
cbo	0

<?php

/**
 * A buffer of items from the queue that can be checked out
 */
class Jetpack_Sync_Queue_Buffer {
	public $id;
	public $items_with_ids;

	public function __construct( $id, $items_with_ids ) {
		$this->id             = $id;
		$this->items_with_ids = $items_with_ids;
	}

	public function get_items() {
		return array_combine( $this->get_item_ids(), $this->get_item_values() );
	}

	public function get_item_values() {
		return Jetpack_Sync_Utils::get_item_values( $this->items_with_ids );
	}

	public function get_item_ids() {
		return Jetpack_Sync_Utils::get_item_ids( $this->items_with_ids );
	}
}

/**
 * A persistent queue that can be flushed in increments of N items,
 * and which blocks reads until checked-out buffers are checked in or
 * closed. This uses raw SQL for two reasons: speed, and not triggering
 * tons of added_option callbacks.
 */
class Jetpack_Sync_Queue {
	public $id;
	private $row_iterator;

	function __construct( $id ) {
		$this->id           = str_replace( '-', '_', $id ); // necessary to ensure we don't have ID collisions in the SQL
		$this->row_iterator = 0;
		$this->random_int   = mt_rand( 1, 1000000 );
class MyClass { }

$x = new MyClass();
$x->foo = true;
	}

	function add( $item ) {
		global $wpdb;
		$added = false;
		// this basically tries to add the option until enough time has elapsed that
		// it has a unique (microtime-based) option key
		while ( ! $added ) {
			$rows_added = $wpdb->query(
				$wpdb->prepare(
					"INSERT INTO $wpdb->options (option_name, option_value, autoload) VALUES (%s, %s,%s)",
					$this->get_next_data_row_option_name(),
					serialize( $item ),
					'no'
				)
			);
			$added      = ( 0 !== $rows_added );
		}
	}

	// Attempts to insert all the items in a single SQL query. May be subject to query size limits!
	function add_all( $items ) {
		global $wpdb;
		$base_option_name = $this->get_next_data_row_option_name();

		$query = "INSERT INTO $wpdb->options (option_name, option_value, autoload) VALUES ";

		$rows = array();

		for ( $i = 0; $i < count( $items ); $i += 1 ) {
for ($i=0; $i<count($array); $i++) { // calls count() on each iteration
}

// Better
for ($i=0, $c=count($array); $i<$c; $i++) { // calls count() just once
}
			$option_name  = esc_sql( $base_option_name . '-' . $i );
			$option_value = esc_sql( serialize( $items[ $i ] ) );
			$rows[]       = "('$option_name', '$option_value', 'no')";
		}

		$rows_added = $wpdb->query( $query . join( ',', $rows ) );

		if ( count( $items ) === $rows_added ) {
			return new WP_Error( 'row_count_mismatch', "The number of rows inserted didn't match the size of the input array" );
		}
	}

	// Peek at the front-most item on the queue without checking it out
	function peek( $count = 1 ) {
		$items = $this->fetch_items( $count );
		if ( $items ) {
			return Jetpack_Sync_Utils::get_item_values( $items );
		}

		return array();
	}

	// lag is the difference in time between the age of the oldest item
	// (aka first or frontmost item) and the current time
	function lag( $now = null ) {
		global $wpdb;

		$first_item_name = $wpdb->get_var(
			$wpdb->prepare(
				"SELECT option_name FROM $wpdb->options WHERE option_name LIKE %s ORDER BY option_name ASC LIMIT 1",
				"jpsq_{$this->id}-%"
			)
		);

		if ( ! $first_item_name ) {
			return 0;
		}

		if ( null === $now ) {
			$now = microtime( true );
		}

		// break apart the item name to get the timestamp
		$matches = null;
		if ( preg_match( '/^jpsq_' . $this->id . '-(\d+\.\d+)-/', $first_item_name, $matches ) ) {
			return $now - floatval( $matches[1] );
		} else {
			return 0;
		}
	}

	function reset() {
		global $wpdb;
		$this->delete_checkout_id();
		$wpdb->query(
			$wpdb->prepare(
				"DELETE FROM $wpdb->options WHERE option_name LIKE %s",
				"jpsq_{$this->id}-%"
			)
		);
	}

	function size() {
		global $wpdb;

		return (int) $wpdb->get_var(
			$wpdb->prepare(
				"SELECT count(*) FROM $wpdb->options WHERE option_name LIKE %s",
				"jpsq_{$this->id}-%"
			)
		);
	}

	// we use this peculiar implementation because it's much faster than count(*)
	function has_any_items() {
		global $wpdb;
		$value = $wpdb->get_var(
			$wpdb->prepare(
				"SELECT exists( SELECT option_name FROM $wpdb->options WHERE option_name LIKE %s )",
				"jpsq_{$this->id}-%"
			)
		);

		return ( $value === '1' );
	}

	function checkout( $buffer_size ) {
		if ( $this->get_checkout_id() ) {
			return new WP_Error( 'unclosed_buffer', 'There is an unclosed buffer' );
		}

		$buffer_id = uniqid();

		$result = $this->set_checkout_id( $buffer_id );

		if ( ! $result || is_wp_error( $result ) ) {
			return $result;
		}

		$items = $this->fetch_items( $buffer_size );

		if ( count( $items ) === 0 ) {
			return false;
		}

		$buffer = new Jetpack_Sync_Queue_Buffer( $buffer_id, array_slice( $items, 0, $buffer_size ) );

		return $buffer;
	}

	// this checks out rows until it either empties the queue or hits a certain memory limit
	// it loads the sizes from the DB first so that it doesn't accidentally
	// load more data into memory than it needs to.
	// The only way it will load more items than $max_size is if a single queue item
	// exceeds the memory limit, but in that case it will send that item by itself.
	function checkout_with_memory_limit( $max_memory, $max_buffer_size = 500 ) {
		if ( $this->get_checkout_id() ) {
			return new WP_Error( 'unclosed_buffer', 'There is an unclosed buffer' );
		}

		$buffer_id = uniqid();

		$result = $this->set_checkout_id( $buffer_id );

		if ( ! $result || is_wp_error( $result ) ) {
			return $result;
		}

		// get the map of buffer_id -> memory_size
		global $wpdb;

		$items_with_size = $wpdb->get_results(
			$wpdb->prepare(
				"SELECT option_name AS id, LENGTH(option_value) AS value_size FROM $wpdb->options WHERE option_name LIKE %s ORDER BY option_name ASC LIMIT %d",
				"jpsq_{$this->id}-%",
				$max_buffer_size
			),
			OBJECT
		);

		if ( count( $items_with_size ) === 0 ) {
			return false;
		}

		$total_memory = 0;

		$min_item_id = $max_item_id = $items_with_size[0]->id;

		foreach ( $items_with_size as $id => $item_with_size ) {
			$total_memory += $item_with_size->value_size;

			// if this is the first item and it exceeds memory, allow loop to continue
			// we will exit on the next iteration instead
			if ( $total_memory > $max_memory && $id > 0 ) {
				break;
			}

			$max_item_id = $item_with_size->id;
		}

		$query = $wpdb->prepare(
			"SELECT option_name AS id, option_value AS value FROM $wpdb->options WHERE option_name >= %s and option_name <= %s ORDER BY option_name ASC",
			$min_item_id,
			$max_item_id
		);

		$items = $wpdb->get_results( $query, OBJECT );
		foreach ( $items as $item ) {
			$item->value = maybe_unserialize( $item->value );
		}

		if ( count( $items ) === 0 ) {
			$this->delete_checkout_id();

			return false;
		}

		$buffer = new Jetpack_Sync_Queue_Buffer( $buffer_id, $items );

		return $buffer;
	}

	function checkin( $buffer ) {
		$is_valid = $this->validate_checkout( $buffer );

		if ( is_wp_error( $is_valid ) ) {
			return $is_valid;
		}

		$this->delete_checkout_id();

		return true;
	}

	function close( $buffer, $ids_to_remove = null ) {
		$is_valid = $this->validate_checkout( $buffer );

		if ( is_wp_error( $is_valid ) ) {
			return $is_valid;
		}

		$this->delete_checkout_id();

		// by default clear all items in the buffer
		if ( is_null( $ids_to_remove ) ) {
			$ids_to_remove = $buffer->get_item_ids();
		}

		global $wpdb;

		if ( count( $ids_to_remove ) > 0 ) {
			$sql   = "DELETE FROM $wpdb->options WHERE option_name IN (" . implode( ', ', array_fill( 0, count( $ids_to_remove ), '%s' ) ) . ')';
			$query = call_user_func_array( array( $wpdb, 'prepare' ), array_merge( array( $sql ), $ids_to_remove ) );
			$wpdb->query( $query );
		}

		return true;
	}

	function flush_all() {
		$items = Jetpack_Sync_Utils::get_item_values( $this->fetch_items() );
		$this->reset();

		return $items;
	}

	function get_all() {
		return $this->fetch_items();
	}

	// use with caution, this could allow multiple processes to delete
	// and send from the queue at the same time
	function force_checkin() {
		$this->delete_checkout_id();
	}

	// used to lock checkouts from the queue.
	// tries to wait up to $timeout seconds for the queue to be empty
	function lock( $timeout = 30 ) {
		$tries = 0;

		while ( $this->has_any_items() && $tries < $timeout ) {
			sleep( 1 );
			$tries += 1;
		}

		if ( $tries === 30 ) {
			return new WP_Error( 'lock_timeout', 'Timeout waiting for sync queue to empty' );
		}

		if ( $this->get_checkout_id() ) {
			return new WP_Error( 'unclosed_buffer', 'There is an unclosed buffer' );
		}

		// hopefully this means we can acquire a checkout?
		$result = $this->set_checkout_id( 'lock' );

		if ( ! $result || is_wp_error( $result ) ) {
			return $result;
		}

		return true;
	}

	function unlock() {
		return $this->delete_checkout_id();
	}

	/**
	 * This option is specifically chosen to, as much as possible, preserve time order
	 * and minimise the possibility of collisions between multiple processes working
	 * at the same time.
	 *
	 * @return string
	 */
	protected function generate_option_name_timestamp() {
		return sprintf( '%.6f', microtime( true ) );
	}

	private function get_checkout_id() {
		global $wpdb;
		$checkout_value = $wpdb->get_var(
			$wpdb->prepare(
				"SELECT option_value FROM $wpdb->options WHERE option_name = %s",
				$this->get_lock_option_name()
			)
		);

		if ( $checkout_value ) {
			list( $checkout_id, $timestamp ) = explode( ':', $checkout_value );
			if ( intval( $timestamp ) > time() ) {
				return $checkout_id;
			}
		}

		return false;
	}

	private function set_checkout_id( $checkout_id ) {
		global $wpdb;

		$expires     = time() + Jetpack_Sync_Defaults::$default_sync_queue_lock_timeout;

		$updated_num = $wpdb->query(
			$wpdb->prepare(
				"UPDATE $wpdb->options SET option_value = %s WHERE option_name = %s",
				"$checkout_id:$expires",
				$this->get_lock_option_name()
			)
		);

		if ( ! $updated_num ) {
			$updated_num = $wpdb->query(
				$wpdb->prepare(
					"INSERT INTO $wpdb->options ( option_name, option_value, autoload ) VALUES ( %s, %s, 'no' )",
					$this->get_lock_option_name(),
					"$checkout_id:$expires"
				)
			);
		}

		return $updated_num;
	}

	private function delete_checkout_id() {
		global $wpdb;
		// rather than delete, which causes fragmentation, we update in place
		return $wpdb->query(
			$wpdb->prepare(
				"UPDATE $wpdb->options SET option_value = %s WHERE option_name = %s",
				'0:0',
				$this->get_lock_option_name()
			)
		);

	}

	private function get_lock_option_name() {
		return "jpsq_{$this->id}_checkout";
	}

	private function get_next_data_row_option_name() {
		$timestamp = $this->generate_option_name_timestamp();

		// row iterator is used to avoid collisions where we're writing data waaay fast in a single process
		if ( $this->row_iterator === PHP_INT_MAX ) {
			$this->row_iterator = 0;
		} else {
			$this->row_iterator += 1;
		}

		return 'jpsq_' . $this->id . '-' . $timestamp . '-' . $this->random_int . '-' . $this->row_iterator;
	}

	private function fetch_items( $limit = null ) {
		global $wpdb;

		if ( $limit ) {
			$query_sql = $wpdb->prepare( "SELECT option_name AS id, option_value AS value FROM $wpdb->options WHERE option_name LIKE %s ORDER BY option_name ASC LIMIT %d", "jpsq_{$this->id}-%", $limit );
		} else {
			$query_sql = $wpdb->prepare( "SELECT option_name AS id, option_value AS value FROM $wpdb->options WHERE option_name LIKE %s ORDER BY option_name ASC", "jpsq_{$this->id}-%" );
		}

		$items = $wpdb->get_results( $query_sql, OBJECT );
		foreach ( $items as $item ) {
			$item->value = maybe_unserialize( $item->value );
		}

		return $items;
	}

	private function validate_checkout( $buffer ) {
		if ( ! $buffer instanceof Jetpack_Sync_Queue_Buffer ) {
			return new WP_Error( 'not_a_buffer', 'You must checkin an instance of Jetpack_Sync_Queue_Buffer' );
		}

		$checkout_id = $this->get_checkout_id();

		if ( ! $checkout_id ) {
			return new WP_Error( 'buffer_not_checked_out', 'There are no checked out buffers' );
		}

		if ( $checkout_id != $buffer->id ) {
			return new WP_Error( 'buffer_mismatch', 'The buffer you checked in was not checked out' );
		}

		return true;
	}
}

class Jetpack_Sync_Utils {

	static function get_item_values( $items ) {
		return array_map( array( __CLASS__, 'get_item_value' ), $items );
	}

	static function get_item_ids( $items ) {
		return array_map( array( __CLASS__, 'get_item_id' ), $items );
	}

	private static function get_item_value( $item ) {
		return $item->value;
	}

	private static function get_item_id( $item ) {
		return $item->id;
	}
}


1			<?php
2
3			/**
4			* A buffer of items from the queue that can be checked out
5			*/
6			class Jetpack_Sync_Queue_Buffer {
7			public $id;
8			public $items_with_ids;
9
10			public function __construct( $id, $items_with_ids ) {
11			$this->id = $id;
12			$this->items_with_ids = $items_with_ids;
13			}
14
15			public function get_items() {
16			return array_combine( $this->get_item_ids(), $this->get_item_values() );
17			}
18
19			public function get_item_values() {
20			return Jetpack_Sync_Utils::get_item_values( $this->items_with_ids );
21			}
22
23			public function get_item_ids() {
24			return Jetpack_Sync_Utils::get_item_ids( $this->items_with_ids );
25			}
26			}
27
28			/**
29			* A persistent queue that can be flushed in increments of N items,
30			* and which blocks reads until checked-out buffers are checked in or
31			* closed. This uses raw SQL for two reasons: speed, and not triggering
32			* tons of added_option callbacks.
33			*/
34			class Jetpack_Sync_Queue {
35			public $id;
36			private $row_iterator;
37
38			function __construct( $id ) {
39			$this->id = str_replace( '-', '_', $id ); // necessary to ensure we don't have ID collisions in the SQL
40			$this->row_iterator = 0;
41			$this->random_int = mt_rand( 1, 1000000 );
			0 ignored issues – show Bug introduced 2016-08-19 19:38 UTC by Report Bug Copy Issue Report The property `random_int` does not exist. Did you maybe forget to declare it? In PHP it is possible to write to properties without declaring them. For example, the following is perfectly valid PHP code: class MyClass { } $x = new MyClass(); $x->foo = true; Generally, it is a good practice to explictly declare properties to avoid accidental typos and provide IDE auto-completion: class MyClass { public $foo; } $x = new MyClass(); $x->foo = true; Loading history...
42			}
43
44			function add( $item ) {
45			global $wpdb;
46			$added = false;
47			// this basically tries to add the option until enough time has elapsed that
48			// it has a unique (microtime-based) option key
49			while ( ! $added ) {
50			$rows_added = $wpdb->query(
51			$wpdb->prepare(
52			"INSERT INTO $wpdb->options (option_name, option_value, autoload) VALUES (%s, %s,%s)",
53			$this->get_next_data_row_option_name(),
54			serialize( $item ),
55			'no'
56			)
57			);
58			$added = ( 0 !== $rows_added );
59			}
60			}
61
62			// Attempts to insert all the items in a single SQL query. May be subject to query size limits!
63			function add_all( $items ) {
64			global $wpdb;
65			$base_option_name = $this->get_next_data_row_option_name();
66
67			$query = "INSERT INTO $wpdb->options (option_name, option_value, autoload) VALUES ";
68
69			$rows = array();
70
71			for ( $i = 0; $i < count( $items ); $i += 1 ) {
			0 ignored issues – show Performance Best Practice introduced 2016-04-17 19:11 UTC by Report Bug Copy Issue Report It seems like you are calling the size function `count()` as part of the test condition. You might want to compute the size beforehand, and not on each iteration. If the size of the collection does not change during the iteration, it is generally a good practice to compute it beforehand, and not on each iteration: for ($i=0; $i<count($array); $i++) { // calls count() on each iteration } // Better for ($i=0, $c=count($array); $i<$c; $i++) { // calls count() just once } Loading history...
72			$option_name = esc_sql( $base_option_name . '-' . $i );
73			$option_value = esc_sql( serialize( $items[ $i ] ) );
74			$rows[] = "('$option_name', '$option_value', 'no')";
75			}
76
77			$rows_added = $wpdb->query( $query . join( ',', $rows ) );
78
79			if ( count( $items ) === $rows_added ) {
80			return new WP_Error( 'row_count_mismatch', "The number of rows inserted didn't match the size of the input array" );
81			}
82			}
83
84			// Peek at the front-most item on the queue without checking it out
85			function peek( $count = 1 ) {
86			$items = $this->fetch_items( $count );
87			if ( $items ) {
88			return Jetpack_Sync_Utils::get_item_values( $items );
89			}
90
91			return array();
92			}
93
94			// lag is the difference in time between the age of the oldest item
95			// (aka first or frontmost item) and the current time
96			function lag( $now = null ) {
97			global $wpdb;
98
99			$first_item_name = $wpdb->get_var(
100			$wpdb->prepare(
101			"SELECT option_name FROM $wpdb->options WHERE option_name LIKE %s ORDER BY option_name ASC LIMIT 1",
102			"jpsq_{$this->id}-%"
103			)
104			);
105
106			if ( ! $first_item_name ) {
107			return 0;
108			}
109
110			if ( null === $now ) {
111			$now = microtime( true );
112			}
113
114			// break apart the item name to get the timestamp
115			$matches = null;
116			if ( preg_match( '/^jpsq_' . $this->id . '-(\d+\.\d+)-/', $first_item_name, $matches ) ) {
117			return $now - floatval( $matches[1] );
118			} else {
119			return 0;
120			}
121			}
122
123			function reset() {
124			global $wpdb;
125			$this->delete_checkout_id();
126			$wpdb->query(
127			$wpdb->prepare(
128			"DELETE FROM $wpdb->options WHERE option_name LIKE %s",
129			"jpsq_{$this->id}-%"
130			)
131			);
132			}
133
134			function size() {
135			global $wpdb;
136
137			return (int) $wpdb->get_var(
138			$wpdb->prepare(
139			"SELECT count(*) FROM $wpdb->options WHERE option_name LIKE %s",
140			"jpsq_{$this->id}-%"
141			)
142			);
143			}
144
145			// we use this peculiar implementation because it's much faster than count(*)
146			function has_any_items() {
147			global $wpdb;
148			$value = $wpdb->get_var(
149			$wpdb->prepare(
150			"SELECT exists( SELECT option_name FROM $wpdb->options WHERE option_name LIKE %s )",
151			"jpsq_{$this->id}-%"
152			)
153			);
154
155			return ( $value === '1' );
156			}
157
158			function checkout( $buffer_size ) {
159			if ( $this->get_checkout_id() ) {
160			return new WP_Error( 'unclosed_buffer', 'There is an unclosed buffer' );
161			}
162
163			$buffer_id = uniqid();
164
165			$result = $this->set_checkout_id( $buffer_id );
166
167			if ( ! $result \|\| is_wp_error( $result ) ) {
168			return $result;
169			}
170
171			$items = $this->fetch_items( $buffer_size );
172
173			if ( count( $items ) === 0 ) {
174			return false;
175			}
176
177			$buffer = new Jetpack_Sync_Queue_Buffer( $buffer_id, array_slice( $items, 0, $buffer_size ) );
178
179			return $buffer;
180			}
181
182			// this checks out rows until it either empties the queue or hits a certain memory limit
183			// it loads the sizes from the DB first so that it doesn't accidentally
184			// load more data into memory than it needs to.
185			// The only way it will load more items than $max_size is if a single queue item
186			// exceeds the memory limit, but in that case it will send that item by itself.
187			function checkout_with_memory_limit( $max_memory, $max_buffer_size = 500 ) {
188			if ( $this->get_checkout_id() ) {
189			return new WP_Error( 'unclosed_buffer', 'There is an unclosed buffer' );
190			}
191
192			$buffer_id = uniqid();
193
194			$result = $this->set_checkout_id( $buffer_id );
195
196			if ( ! $result \|\| is_wp_error( $result ) ) {
197			return $result;
198			}
199
200			// get the map of buffer_id -> memory_size
201			global $wpdb;
202
203			$items_with_size = $wpdb->get_results(
204			$wpdb->prepare(
205			"SELECT option_name AS id, LENGTH(option_value) AS value_size FROM $wpdb->options WHERE option_name LIKE %s ORDER BY option_name ASC LIMIT %d",
206			"jpsq_{$this->id}-%",
207			$max_buffer_size
208			),
209			OBJECT
210			);
211
212			if ( count( $items_with_size ) === 0 ) {
213			return false;
214			}
215
216			$total_memory = 0;
217
218			$min_item_id = $max_item_id = $items_with_size[0]->id;
219
220			foreach ( $items_with_size as $id => $item_with_size ) {
221			$total_memory += $item_with_size->value_size;
222
223			// if this is the first item and it exceeds memory, allow loop to continue
224			// we will exit on the next iteration instead
225			if ( $total_memory > $max_memory && $id > 0 ) {
226			break;
227			}
228
229			$max_item_id = $item_with_size->id;
230			}
231
232			$query = $wpdb->prepare(
233			"SELECT option_name AS id, option_value AS value FROM $wpdb->options WHERE option_name >= %s and option_name <= %s ORDER BY option_name ASC",
234			$min_item_id,
235			$max_item_id
236			);
237
238			$items = $wpdb->get_results( $query, OBJECT );
239			foreach ( $items as $item ) {
240			$item->value = maybe_unserialize( $item->value );
241			}
242
243			if ( count( $items ) === 0 ) {
244			$this->delete_checkout_id();
245
246			return false;
247			}
248
249			$buffer = new Jetpack_Sync_Queue_Buffer( $buffer_id, $items );
250
251			return $buffer;
252			}
253
254			function checkin( $buffer ) {
255			$is_valid = $this->validate_checkout( $buffer );
256
257			if ( is_wp_error( $is_valid ) ) {
258			return $is_valid;
259			}
260
261			$this->delete_checkout_id();
262
263			return true;
264			}
265
266			function close( $buffer, $ids_to_remove = null ) {
267			$is_valid = $this->validate_checkout( $buffer );
268
269			if ( is_wp_error( $is_valid ) ) {
270			return $is_valid;
271			}
272
273			$this->delete_checkout_id();
274
275			// by default clear all items in the buffer
276			if ( is_null( $ids_to_remove ) ) {
277			$ids_to_remove = $buffer->get_item_ids();
278			}
279
280			global $wpdb;
281
282			if ( count( $ids_to_remove ) > 0 ) {
283			$sql = "DELETE FROM $wpdb->options WHERE option_name IN (" . implode( ', ', array_fill( 0, count( $ids_to_remove ), '%s' ) ) . ')';
284			$query = call_user_func_array( array( $wpdb, 'prepare' ), array_merge( array( $sql ), $ids_to_remove ) );
285			$wpdb->query( $query );
286			}
287
288			return true;
289			}
290
291			function flush_all() {
292			$items = Jetpack_Sync_Utils::get_item_values( $this->fetch_items() );
293			$this->reset();
294
295			return $items;
296			}
297
298			function get_all() {
299			return $this->fetch_items();
300			}
301
302			// use with caution, this could allow multiple processes to delete
303			// and send from the queue at the same time
304			function force_checkin() {
305			$this->delete_checkout_id();
306			}
307
308			// used to lock checkouts from the queue.
309			// tries to wait up to $timeout seconds for the queue to be empty
310			function lock( $timeout = 30 ) {
311			$tries = 0;
312
313			while ( $this->has_any_items() && $tries < $timeout ) {
314			sleep( 1 );
315			$tries += 1;
316			}
317
318			if ( $tries === 30 ) {
319			return new WP_Error( 'lock_timeout', 'Timeout waiting for sync queue to empty' );
320			}
321
322			if ( $this->get_checkout_id() ) {
323			return new WP_Error( 'unclosed_buffer', 'There is an unclosed buffer' );
324			}
325
326			// hopefully this means we can acquire a checkout?
327			$result = $this->set_checkout_id( 'lock' );
328
329			if ( ! $result \|\| is_wp_error( $result ) ) {
330			return $result;
331			}
332
333			return true;
334			}
335
336			function unlock() {
337			return $this->delete_checkout_id();
338			}
339
340			/**
341			* This option is specifically chosen to, as much as possible, preserve time order
342			* and minimise the possibility of collisions between multiple processes working
343			* at the same time.
344			*
345			* @return string
346			*/
347			protected function generate_option_name_timestamp() {
348			return sprintf( '%.6f', microtime( true ) );
349			}
350
351			private function get_checkout_id() {
352			global $wpdb;
353			$checkout_value = $wpdb->get_var(
354			$wpdb->prepare(
355			"SELECT option_value FROM $wpdb->options WHERE option_name = %s",
356			$this->get_lock_option_name()
357			)
358			);
359
360			if ( $checkout_value ) {
361			list( $checkout_id, $timestamp ) = explode( ':', $checkout_value );
362			if ( intval( $timestamp ) > time() ) {
363			return $checkout_id;
364			}
365			}
366
367			return false;
368			}
369
370			private function set_checkout_id( $checkout_id ) {
371			global $wpdb;
372
373			$expires = time() + Jetpack_Sync_Defaults::$default_sync_queue_lock_timeout;
			0 ignored issues – show Bug introduced 2017-04-18 21:07 UTC by Report Bug Copy Issue Report The property `default_sync_queue_lock_timeout` cannot be accessed from this context as it is declared `private` in class `Jetpack_Sync_Defaults`. This check looks for access to properties that are not accessible from the current context. If you need to make a property accessible to another context you can either raise its visibility level or provide an accessible getter in the defining class. Loading history...
374			$updated_num = $wpdb->query(
375			$wpdb->prepare(
376			"UPDATE $wpdb->options SET option_value = %s WHERE option_name = %s",
377			"$checkout_id:$expires",
378			$this->get_lock_option_name()
379			)
380			);
381
382			if ( ! $updated_num ) {
383			$updated_num = $wpdb->query(
384			$wpdb->prepare(
385			"INSERT INTO $wpdb->options ( option_name, option_value, autoload ) VALUES ( %s, %s, 'no' )",
386			$this->get_lock_option_name(),
387			"$checkout_id:$expires"
388			)
389			);
390			}
391
392			return $updated_num;
393			}
394
395			private function delete_checkout_id() {
396			global $wpdb;
397			// rather than delete, which causes fragmentation, we update in place
398			return $wpdb->query(
399			$wpdb->prepare(
400			"UPDATE $wpdb->options SET option_value = %s WHERE option_name = %s",
401			'0:0',
402			$this->get_lock_option_name()
403			)
404			);
405
406			}
407
408			private function get_lock_option_name() {
409			return "jpsq_{$this->id}_checkout";
410			}
411
412			private function get_next_data_row_option_name() {
413			$timestamp = $this->generate_option_name_timestamp();
414
415			// row iterator is used to avoid collisions where we're writing data waaay fast in a single process
416			if ( $this->row_iterator === PHP_INT_MAX ) {
417			$this->row_iterator = 0;
418			} else {
419			$this->row_iterator += 1;
420			}
421
422			return 'jpsq_' . $this->id . '-' . $timestamp . '-' . $this->random_int . '-' . $this->row_iterator;
423			}
424
425			private function fetch_items( $limit = null ) {
426			global $wpdb;
427
428			if ( $limit ) {
429			$query_sql = $wpdb->prepare( "SELECT option_name AS id, option_value AS value FROM $wpdb->options WHERE option_name LIKE %s ORDER BY option_name ASC LIMIT %d", "jpsq_{$this->id}-%", $limit );
430			} else {
431			$query_sql = $wpdb->prepare( "SELECT option_name AS id, option_value AS value FROM $wpdb->options WHERE option_name LIKE %s ORDER BY option_name ASC", "jpsq_{$this->id}-%" );
432			}
433
434			$items = $wpdb->get_results( $query_sql, OBJECT );
435			foreach ( $items as $item ) {
436			$item->value = maybe_unserialize( $item->value );
437			}
438
439			return $items;
440			}
441
442			private function validate_checkout( $buffer ) {
443			if ( ! $buffer instanceof Jetpack_Sync_Queue_Buffer ) {
444			return new WP_Error( 'not_a_buffer', 'You must checkin an instance of Jetpack_Sync_Queue_Buffer' );
445			}
446
447			$checkout_id = $this->get_checkout_id();
448
449			if ( ! $checkout_id ) {
450			return new WP_Error( 'buffer_not_checked_out', 'There are no checked out buffers' );
451			}
452
453			if ( $checkout_id != $buffer->id ) {
454			return new WP_Error( 'buffer_mismatch', 'The buffer you checked in was not checked out' );
455			}
456
457			return true;
458			}
459			}
460
461			class Jetpack_Sync_Utils {
462
463			static function get_item_values( $items ) {
464			return array_map( array( __CLASS__, 'get_item_value' ), $items );
465			}
466
467			static function get_item_ids( $items ) {
468			return array_map( array( __CLASS__, 'get_item_id' ), $items );
469			}
470
471			private static function get_item_value( $item ) {
472			return $item->value;
473			}
474
475			private static function get_item_id( $item ) {
476			return $item->id;
477			}
478			}
479

Automattic / jetpack

Push — fix/vvv-tests-wp-5-2 ( ec7a8b...e4b178 )

Abstract_Jetpack_Sync_Queue A

Complexity

Size/Duplication

Coupling/Cohesion

Importance

Duplication Side-by-Side

Filter issues like