+global\Jetpack_Sync_Sender::do_sync_for_queue()

Complexity

Conditions	21
Paths	245

Size

Total Lines	172
Code Lines	72

Duplication

Lines	0
Ratio	0 %

Importance

Changes	8
Bugs	2	Features	2

<?php
	public function do_sync_for_queue( $queue ) {

		do_action( 'jetpack_sync_before_send_queue_' . $queue->id );

		if ( $queue->size() === 0 ) {
			return false;
		}

		// now that we're sure we are about to sync, try to
		// ignore user abort so we can avoid getting into a
		// bad state
		if ( function_exists( 'ignore_user_abort' ) ) {
			ignore_user_abort( true );
		}

		$buffer = $queue->checkout_with_memory_limit( $this->dequeue_max_bytes, $this->upload_max_rows );

		if ( ! $buffer ) {
			// buffer has no items
			return false;
		}

		if ( is_wp_error( $buffer ) ) {
			// another buffer is currently sending
			return false;
		}

		$upload_size   = 0;
		$items_to_send = array();
		$items         = $buffer->get_items();

		// set up current screen to avoid errors rendering content
		require_once(ABSPATH . 'wp-admin/includes/class-wp-screen.php');
		require_once(ABSPATH . 'wp-admin/includes/screen.php');
		set_current_screen( 'sync' );

		$skipped_items_ids = array();

		// we estimate the total encoded size as we go by encoding each item individually
		// this is expensive, but the only way to really know :/
		foreach ( $items as $key => $item ) {
			// Suspending cache addition help prevent overloading in memory cache of large sites.
			wp_suspend_cache_addition( true );
			/**
			 * Modify the data within an action before it is serialized and sent to the server
			 * For example, during full sync this expands Post ID's into full Post objects,
			 * so that we don't have to serialize the whole object into the queue.
			 *
			 * @since 4.2.0
			 *
			 * @param array The action parameters
			 * @param int The ID of the user who triggered the action
			 */
			$item_value = apply_filters( 'jetpack_sync_before_send_' . $item[0], $item[1], $item[2] );
			wp_suspend_cache_addition( false );
			if ( $item_value === false ) {
				$skipped_items_ids[] = $key;
				continue;
			}
			if ( $item_value instanceof Traversable ) {
				foreach ( $item_value as $value ) {
					$new_item[0] = $item[0];

$new_item was never initialized. Although not strictly required by PHP, it is generally a good practice to add $new_item = array(); before regardless.

					$new_item[1] = apply_filters( 'jetpack_sync_before_send_' . $item[0], array( $value ), $item[2] );

The variable $new_item does not seem to be defined for all execution paths leading up to this point.

					$new_item[2] = $item[2];
					$new_item[3] = $item[3];
					$new_item[4] = $item[4];

					$encoded_item = $this->codec->encode( $new_item );

					$upload_size += strlen( $encoded_item );

					if ( $upload_size > $this->upload_max_bytes && count( $items_to_send ) > 0 ) {
						break 2;
					}

					$last_item = end( $value );
					$items_to_send[ $key . '-from-' . $last_item ] = $encoded_item;
				}
			} else {

				$item[1] = $item_value;

				$encoded_item = $this->codec->encode( $item );

				$upload_size += strlen( $encoded_item );

				if ( $upload_size > $this->upload_max_bytes && count( $items_to_send ) > 0 ) {
					break;
				}

				$items_to_send[ $key ] = $encoded_item;
			}
		}

		/**
		 * Fires when data is ready to send to the server.
		 * Return false or WP_Error to abort the sync (e.g. if there's an error)
		 * The items will be automatically re-sent later
		 *
		 * @since 4.2.0
		 *
		 * @param array $data The action buffer
		 * @param string $codec The codec name used to encode the data
		 * @param double $time The current time
		 * @param string $queue The queue used to send ('sync' or 'full_sync')
		 */
		$processed_item_ids = apply_filters( 'jetpack_sync_send_data', $items_to_send, $this->codec->name(), microtime( true ), $queue->id );

		if ( ! $processed_item_ids || is_wp_error( $processed_item_ids ) ) {
			$checked_in_item_ids = $queue->checkin( $buffer );

			if ( is_wp_error( $checked_in_item_ids ) ) {
				error_log( 'Error checking in buffer: ' . $checked_in_item_ids->get_error_message() );
				$queue->force_checkin();
			}

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

			// returning a WP_Error is a sign to the caller that we should wait a while
			// before syncing again
			return new WP_Error( 'server_error' );
			
		} else {

			// detect if the last item ID was an error
			$had_wp_error = is_wp_error( end( $processed_item_ids ) );

			if ( $had_wp_error ) {
				$wp_error = array_pop( $processed_item_ids );
			}

			$items_to_update = array();
			$last_processed_item_id = end( $processed_item_ids );

			if ( preg_match( '/^jpsq_full_sync-(\d+\.\d+)-(\d+)-(\d+)-(\d+)$/', $last_processed_item_id, $matches ) ) {
				// detect if it was an item that we expanded during sending
				$parent_item_key = chop( $last_processed_item_id, '-' . $matches[4] );
				$items_to_update[ $parent_item_key ] = $items[ $parent_item_key ];
				$items_to_update[ $parent_item_key ][1] = $matches[4];
			}

			// also checkin any items that were skipped
			if ( count( $skipped_items_ids ) > 0 ) {
				$processed_item_ids = array_merge( $processed_item_ids, $skipped_items_ids );
			}

			$processed_items = array_intersect_key( $items, array_flip( $processed_item_ids ) );

			/**
			 * Allows us to keep track of all the actions that have been sent.
			 * Allows us to calculate the progress of specific actions.
			 *
			 * @since 4.2.0
			 *
			 * @param array $processed_actions The actions that we send successfully.
			 */
			do_action( 'jetpack_sync_processed_actions', $processed_items );

			$queue->close( $buffer, $processed_item_ids, $items_to_update );

			// returning a WP_Error is a sign to the caller that we should wait a while
			// before syncing again
			if ( $had_wp_error ) {
				return $wp_error;

The variable $wp_error does not seem to be defined for all execution paths leading up to this point.

			} 
		}
		
		return true;
	}