Jetpack_Sync_Queue::add_all() - Code Metrics - Inspection of "Fix incorrect import" - Automattic/jetpack - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Push — add/simple-payments-metrics ( 81090c...f7b933 )

unknown

created 2018-06-25 15:28 UTC

Jetpack_Sync_Queue::add_all() A

↳ Parent: Jetpack_Sync_Queue

Complexity

Conditions	3
Paths	4

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	3
nc	4
nop	1
dl	0
loc	20
rs	9.6
c	0
b	0
f	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 );
	}

	static private function get_item_value( $item ) {

		return $item->value;
	}

	static private 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 {
			0 ignored issues – show Coding Style Compatibility introduced 2017-04-18 21:07 UTC by Report Bug Copy Issue Report PSR1 recommends that each class should be in its own file to aid autoloaders. Having each class in a dedicated file usually plays nice with PSR autoloaders and is therefore a well established practice. If you use other autoloaders, you might not want to follow this rule. Loading history...
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( $wpdb->prepare(
51			"INSERT INTO $wpdb->options (option_name, option_value, autoload) VALUES (%s, %s,%s)",
52			$this->get_next_data_row_option_name(),
53			serialize( $item ),
54			'no'
55			) );
56			$added = ( 0 !== $rows_added );
57			}
58			}
59
60			// Attempts to insert all the items in a single SQL query. May be subject to query size limits!
61			function add_all( $items ) {
62			global $wpdb;
63			$base_option_name = $this->get_next_data_row_option_name();
64
65			$query = "INSERT INTO $wpdb->options (option_name, option_value, autoload) VALUES ";
66
67			$rows = array();
68
69			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...
70			$option_name = esc_sql( $base_option_name . '-' . $i );
71			$option_value = esc_sql( serialize( $items[ $i ] ) );
72			$rows[] = "('$option_name', '$option_value', 'no')";
73			}
74
75			$rows_added = $wpdb->query( $query . join( ',', $rows ) );
76
77			if ( count( $items ) === $rows_added ) {
78			return new WP_Error( 'row_count_mismatch', "The number of rows inserted didn't match the size of the input array" );
79			}
80			}
81
82			// Peek at the front-most item on the queue without checking it out
83			function peek( $count = 1 ) {
84			$items = $this->fetch_items( $count );
85			if ( $items ) {
86			return Jetpack_Sync_Utils::get_item_values( $items );
87			}
88
89			return array();
90			}
91
92			// lag is the difference in time between the age of the oldest item
93			// (aka first or frontmost item) and the current time
94			function lag( $now = null ) {
95			global $wpdb;
96
97			$first_item_name = $wpdb->get_var( $wpdb->prepare(
98			"SELECT option_name FROM $wpdb->options WHERE option_name LIKE %s ORDER BY option_name ASC LIMIT 1",
99			"jpsq_{$this->id}-%"
100			) );
101
102			if ( ! $first_item_name ) {
103			return 0;
104			}
105
106			if ( null === $now ) {
107			$now = microtime( true );
108			}
109
110			// break apart the item name to get the timestamp
111			$matches = null;
112			if ( preg_match( '/^jpsq_' . $this->id . '-(\d+\.\d+)-/', $first_item_name, $matches ) ) {
113			return $now - floatval( $matches[1] );
114			} else {
115			return 0;
116			}
117			}
118
119			function reset() {
120			global $wpdb;
121			$this->delete_checkout_id();
122			$wpdb->query( $wpdb->prepare(
123			"DELETE FROM $wpdb->options WHERE option_name LIKE %s", "jpsq_{$this->id}-%"
124			) );
125			}
126
127			function size() {
128			global $wpdb;
129
130			return (int) $wpdb->get_var( $wpdb->prepare(
131			"SELECT count(*) FROM $wpdb->options WHERE option_name LIKE %s", "jpsq_{$this->id}-%"
132			) );
133			}
134
135			// we use this peculiar implementation because it's much faster than count(*)
136			function has_any_items() {
137			global $wpdb;
138			$value = $wpdb->get_var( $wpdb->prepare(
139			"SELECT exists( SELECT option_name FROM $wpdb->options WHERE option_name LIKE %s )", "jpsq_{$this->id}-%"
140			) );
141
142			return ( $value === '1' );
143			}
144
145			function checkout( $buffer_size ) {
146			if ( $this->get_checkout_id() ) {
147			return new WP_Error( 'unclosed_buffer', 'There is an unclosed buffer' );
148			}
149
150			$buffer_id = uniqid();
151
152			$result = $this->set_checkout_id( $buffer_id );
153
154			if ( ! $result \|\| is_wp_error( $result ) ) {
155			return $result;
156			}
157
158			$items = $this->fetch_items( $buffer_size );
159
160			if ( count( $items ) === 0 ) {
161			return false;
162			}
163
164			$buffer = new Jetpack_Sync_Queue_Buffer( $buffer_id, array_slice( $items, 0, $buffer_size ) );
165
166			return $buffer;
167			}
168
169			// this checks out rows until it either empties the queue or hits a certain memory limit
170			// it loads the sizes from the DB first so that it doesn't accidentally
171			// load more data into memory than it needs to.
172			// The only way it will load more items than $max_size is if a single queue item
173			// exceeds the memory limit, but in that case it will send that item by itself.
174			function checkout_with_memory_limit( $max_memory, $max_buffer_size = 500 ) {
175			if ( $this->get_checkout_id() ) {
176			return new WP_Error( 'unclosed_buffer', 'There is an unclosed buffer' );
177			}
178
179			$buffer_id = uniqid();
180
181			$result = $this->set_checkout_id( $buffer_id );
182
183			if ( ! $result \|\| is_wp_error( $result ) ) {
184			return $result;
185			}
186
187			// get the map of buffer_id -> memory_size
188			global $wpdb;
189
190			$items_with_size = $wpdb->get_results(
191			$wpdb->prepare(
192			"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",
193			"jpsq_{$this->id}-%",
194			$max_buffer_size
195			),
196			OBJECT
197			);
198
199			if ( count( $items_with_size ) === 0 ) {
200			return false;
201			}
202
203			$total_memory = 0;
204
205			$min_item_id = $max_item_id = $items_with_size[0]->id;
206
207			foreach ( $items_with_size as $id => $item_with_size ) {
208			$total_memory += $item_with_size->value_size;
209
210			// if this is the first item and it exceeds memory, allow loop to continue
211			// we will exit on the next iteration instead
212			if ( $total_memory > $max_memory && $id > 0 ) {
213			break;
214			}
215
216			$max_item_id = $item_with_size->id;
217			}
218
219			$query = $wpdb->prepare(
220			"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",
221			$min_item_id,
222			$max_item_id
223			);
224
225			$items = $wpdb->get_results( $query, OBJECT );
226			foreach ( $items as $item ) {
227			$item->value = maybe_unserialize( $item->value );
228			}
229
230			if ( count( $items ) === 0 ) {
231			$this->delete_checkout_id();
232
233			return false;
234			}
235
236			$buffer = new Jetpack_Sync_Queue_Buffer( $buffer_id, $items );
237
238			return $buffer;
239			}
240
241			function checkin( $buffer ) {
242			$is_valid = $this->validate_checkout( $buffer );
243
244			if ( is_wp_error( $is_valid ) ) {
245			return $is_valid;
246			}
247
248			$this->delete_checkout_id();
249
250			return true;
251			}
252
253			function close( $buffer, $ids_to_remove = null ) {
254			$is_valid = $this->validate_checkout( $buffer );
255
256			if ( is_wp_error( $is_valid ) ) {
257			return $is_valid;
258			}
259
260			$this->delete_checkout_id();
261
262			// by default clear all items in the buffer
263			if ( is_null( $ids_to_remove ) ) {
264			$ids_to_remove = $buffer->get_item_ids();
265			}
266
267			global $wpdb;
268
269			if ( count( $ids_to_remove ) > 0 ) {
270			$sql = "DELETE FROM $wpdb->options WHERE option_name IN (" . implode( ', ', array_fill( 0, count( $ids_to_remove ), '%s' ) ) . ')';
271			$query = call_user_func_array( array( $wpdb, 'prepare' ), array_merge( array( $sql ), $ids_to_remove ) );
272			$wpdb->query( $query );
273			}
274
275			return true;
276			}
277
278			function flush_all() {
279			$items = Jetpack_Sync_Utils::get_item_values( $this->fetch_items() );
280			$this->reset();
281
282			return $items;
283			}
284
285			function get_all() {
286			return $this->fetch_items();
287			}
288
289			// use with caution, this could allow multiple processes to delete
290			// and send from the queue at the same time
291			function force_checkin() {
292			$this->delete_checkout_id();
293			}
294
295			// used to lock checkouts from the queue.
296			// tries to wait up to $timeout seconds for the queue to be empty
297			function lock( $timeout = 30 ) {
298			$tries = 0;
299
300			while ( $this->has_any_items() && $tries < $timeout ) {
301			sleep( 1 );
302			$tries += 1;
303			}
304
305			if ( $tries === 30 ) {
306			return new WP_Error( 'lock_timeout', 'Timeout waiting for sync queue to empty' );
307			}
308
309			if ( $this->get_checkout_id() ) {
310			return new WP_Error( 'unclosed_buffer', 'There is an unclosed buffer' );
311			}
312
313			// hopefully this means we can acquire a checkout?
314			$result = $this->set_checkout_id( 'lock' );
315
316			if ( ! $result \|\| is_wp_error( $result ) ) {
317			return $result;
318			}
319
320			return true;
321			}
322
323			function unlock() {
324			return $this->delete_checkout_id();
325			}
326
327			/**
328			* This option is specifically chosen to, as much as possible, preserve time order
329			* and minimise the possibility of collisions between multiple processes working
330			* at the same time.
331			*
332			* @return string
333			*/
334			protected function generate_option_name_timestamp() {
335			return sprintf( '%.6f', microtime( true ) );
336			}
337
338			private function get_checkout_id() {
339			global $wpdb;
340			$checkout_value = $wpdb->get_var(
341			$wpdb->prepare(
342			"SELECT option_value FROM $wpdb->options WHERE option_name = %s",
343			$this->get_lock_option_name()
344			)
345			);
346
347			if ( $checkout_value ) {
348			list( $checkout_id, $timestamp ) = explode( ':', $checkout_value );
349			if ( intval( $timestamp ) > time() ) {
350			return $checkout_id;
351			}
352			}
353
354			return false;
355			}
356
357			private function set_checkout_id( $checkout_id ) {
358			global $wpdb;
359
360			$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...
361			$updated_num = $wpdb->query(
362			$wpdb->prepare(
363			"UPDATE $wpdb->options SET option_value = %s WHERE option_name = %s",
364			"$checkout_id:$expires",
365			$this->get_lock_option_name()
366			)
367			);
368
369			if ( ! $updated_num ) {
370			$updated_num = $wpdb->query(
371			$wpdb->prepare(
372			"INSERT INTO $wpdb->options ( option_name, option_value, autoload ) VALUES ( %s, %s, 'no' )",
373			$this->get_lock_option_name(),
374			"$checkout_id:$expires"
375			)
376			);
377			}
378
379			return $updated_num;
380			}
381
382			private function delete_checkout_id() {
383			global $wpdb;
384			// rather than delete, which causes fragmentation, we update in place
385			return $wpdb->query(
386			$wpdb->prepare(
387			"UPDATE $wpdb->options SET option_value = %s WHERE option_name = %s",
388			"0:0",
389			$this->get_lock_option_name()
390			)
391			);
392
393			}
394
395			private function get_lock_option_name() {
396			return "jpsq_{$this->id}_checkout";
397			}
398
399			private function get_next_data_row_option_name() {
400			$timestamp = $this->generate_option_name_timestamp();
401
402			// row iterator is used to avoid collisions where we're writing data waaay fast in a single process
403			if ( $this->row_iterator === PHP_INT_MAX ) {
404			$this->row_iterator = 0;
405			} else {
406			$this->row_iterator += 1;
407			}
408
409			return 'jpsq_' . $this->id . '-' . $timestamp . '-' . $this->random_int . '-' . $this->row_iterator;
410			}
411
412			private function fetch_items( $limit = null ) {
413			global $wpdb;
414
415			if ( $limit ) {
416			$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 );
417			} else {
418			$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}-%" );
419			}
420
421			$items = $wpdb->get_results( $query_sql, OBJECT );
422			foreach ( $items as $item ) {
423			$item->value = maybe_unserialize( $item->value );
424			}
425
426			return $items;
427			}
428
429			private function validate_checkout( $buffer ) {
430			if ( ! $buffer instanceof Jetpack_Sync_Queue_Buffer ) {
431			return new WP_Error( 'not_a_buffer', 'You must checkin an instance of Jetpack_Sync_Queue_Buffer' );
432			}
433
434			$checkout_id = $this->get_checkout_id();
435
436			if ( ! $checkout_id ) {
437			return new WP_Error( 'buffer_not_checked_out', 'There are no checked out buffers' );
438			}
439
440			if ( $checkout_id != $buffer->id ) {
441			return new WP_Error( 'buffer_mismatch', 'The buffer you checked in was not checked out' );
442			}
443
444			return true;
445			}
446			}
447
448			class Jetpack_Sync_Utils {
			0 ignored issues – show Coding Style Compatibility introduced 2016-04-19 21:52 UTC by Report Bug Copy Issue Report PSR1 recommends that each class should be in its own file to aid autoloaders. Having each class in a dedicated file usually plays nice with PSR autoloaders and is therefore a well established practice. If you use other autoloaders, you might not want to follow this rule. Loading history...
449
450			static function get_item_values( $items ) {
451			return array_map( array( __CLASS__, 'get_item_value' ), $items );
452			}
453
454			static function get_item_ids( $items ) {
455			return array_map( array( __CLASS__, 'get_item_id' ), $items );
456			}
457
458			static private function get_item_value( $item ) {
			0 ignored issues – show Coding Style introduced 2017-07-31 16:04 UTC by Report Bug Copy Issue Report As per PSR2, the `static` declaration should come after the visibility declaration. Loading history...
459			return $item->value;
460			}
461
462			static private function get_item_id( $item ) {
			0 ignored issues – show Coding Style introduced 2017-07-31 16:04 UTC by Report Bug Copy Issue Report As per PSR2, the `static` declaration should come after the visibility declaration. Loading history...
463			return $item->id;
464			}
465			}
466

Automattic / jetpack

Push — add/simple-payments-metrics ( 81090c...f7b933 )

Jetpack_Sync_Queue::add_all() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like