xero_sync-8.x-1.x-dev/src/XeroSyncQueueJobTrait.php
src/XeroSyncQueueJobTrait.php
<?php
namespace Drupal\xero_sync;
use Drupal\Component\Datetime\TimeInterface;
use Drupal\Component\Plugin\Exception\InvalidPluginDefinitionException;
use Drupal\Component\Plugin\Exception\PluginNotFoundException;
use Drupal\Component\Render\FormattableMarkup;
use Drupal\Core\Entity\EntityTypeManagerInterface;
use Drupal\Core\Entity\EntityTypeBundleInfoInterface;
use Drupal\Core\TypedData\TypedDataManagerInterface;
use Drupal\xero\Exception\XeroInvalidConfigurationException;
use Drupal\xero_sync\Exception\SyncFailureException;
use Drupal\xero_sync\Exception\SyncRetryException;
use GuzzleHttp\Exception\RequestException;
use Psr\Log\LoggerInterface;
use Drupal\Core\Logger\RfcLogLevel;
use Symfony\Component\DependencyInjection\ContainerInterface;
use Drupal\Core\Entity\EntityInterface;
use Symfony\Component\Serializer\SerializerInterface;
/**
* A trait for sharing between QueueWorker and AdvancedQueue/JobType plugins.
*/
trait XeroSyncQueueJobTrait {
use XeroSyncUtilitiesTrait;
/**
* A logger instance.
*
* @var \Psr\Log\LoggerInterface
*/
protected $logger;
/**
* The entity type manager.
*
* @var \Drupal\Core\Entity\EntityTypeManagerInterface
*/
protected $entityTypeManager;
/**
* The entity type bundle manager.
*
* @var \Drupal\Core\Entity\EntityTypeBundleInfoInterface
*/
protected $entityTypeBundleInfo;
/**
* The typed data manager service.
*
* @var \Drupal\Core\TypedData\TypedDataManagerInterface
*/
protected $typedDataManager;
/**
* The job manager service.
*
* @var \Drupal\xero_sync\XeroSyncJobManagerInterface
*/
protected $jobManager;
/**
* The serializer.
*
* @var \Symfony\Component\Serializer\SerializerInterface
*/
protected $serializer;
/**
* The entity handler service.
*
* @var \Drupal\xero_sync\XeroSyncEntityHandlerInterface
*/
protected $entityHandler;
/**
* Set properties on the class, as part of constructing the object.
*
* Having the full construct method on the trait makes unit testing harder.
*
* @param \Psr\Log\LoggerInterface $logger
* A logger instance.
* @param \Drupal\Core\Entity\EntityTypeManagerInterface $entity_type_manager
* The entity type manager.
* @param \Drupal\Core\Entity\EntityTypeBundleInfoInterface $entity_type_bundle_info
* The entity type bundle manager.
* @param \Drupal\xero_sync\XeroSyncSynchronizerInterface $synchronizer
* The synchronizer.
* @param \Drupal\Core\TypedData\TypedDataManagerInterface $typed_data_manager
* The typed data manager service.
* @param \Drupal\xero_sync\XeroSyncJobManagerInterface $job_manager
* The job manager.
* @param \Symfony\Component\Serializer\SerializerInterface $serializer
* The serializer.
* @param \Drupal\Component\Datetime\TimeInterface $time
* The time service.
* @param \Drupal\xero_sync\XeroSyncEntityHandlerInterface $entity_handler
* The entity handler.
*/
protected function setProperties(LoggerInterface $logger, EntityTypeManagerInterface $entity_type_manager, EntityTypeBundleInfoInterface $entity_type_bundle_info, XeroSyncSynchronizerInterface $synchronizer, TypedDataManagerInterface $typed_data_manager, XeroSyncJobManagerInterface $job_manager, SerializerInterface $serializer, TimeInterface $time, XeroSyncEntityHandlerInterface $entity_handler) {
$this->logger = $logger;
$this->entityTypeManager = $entity_type_manager;
$this->entityTypeBundleInfo = $entity_type_bundle_info;
$this->synchronizer = $synchronizer;
$this->typedDataManager = $typed_data_manager;
$this->jobManager = $job_manager;
$this->serializer = $serializer;
$this->time = $time;
$this->entityHandler = $entity_handler;
}
/**
* {@inheritdoc}
*/
public static function create(ContainerInterface $container, array $configuration, $plugin_id, $plugin_definition) {
return new static(
$configuration,
$plugin_id,
$plugin_definition,
$container->get('logger.channel.xero_sync'),
$container->get('entity_type.manager'),
$container->get('entity_type.bundle.info'),
$container->get('xero_sync.synchronizer'),
$container->get('typed_data_manager'),
$container->get('xero_sync.job_manager'),
$container->get('serializer'),
$container->get('datetime.time'),
$container->get('xero_sync.entity_handler')
);
}
/**
* Process a queue job item, logging failures.
*
* @param array $data
* The job data.
* @param string $queueId
* (optional) The queue id.
*/
protected function doProcessItem(array $data, $queueId = NULL) {
// When processing queue items,we get the queue id from the plugin id.
// But we allow it to be set explicitly to make testing easier.
if (empty($queueId)) {
$queueId = $this->getPluginId();
}
if (!isset($data['entity_type'])) {
$message = (string) new FormattableMarkup('No entity type specified using "entity_type" key when processing a job in queue @queue_id. \n @data ', [
'@queue_id' => $queueId,
'@data' => print_r($data, TRUE),
]);
$this->logger->log(RfcLogLevel::ERROR, $message);
throw new \InvalidArgumentException($message);
}
if (isset($data['entity'])) {
$entity_class = NULL;
if (isset($data['entity_bundle'])) {
// @todo #2570593 may provide a service method to help with this.
$bundle_info = $this->entityTypeBundleInfo->getBundleInfo($data['entity_type']);
$entity_class = $bundle_info[$data['entity_bundle']]['class'] ?? NULL;
}
if (is_null($entity_class)) {
try {
$entityType = $this->entityTypeManager->getDefinition($data['entity_type']);
}
catch (PluginNotFoundException $e) {
$message = (string) new FormattableMarkup('Invalid entity type specified using "entity_type" key when processing a job in queue @queue_id. \n @data ', [
'@queue_id' => $queueId,
'@data' => print_r($data, TRUE),
]);
$this->logger->log(RfcLogLevel::ERROR, $message);
throw new \InvalidArgumentException($message);
}
$entity_class = $entityType->getClass();
}
$entity = $this->serializer->deserialize($data['entity'], $entity_class, 'json');
}
else {
try {
$entityStorage = $this->entityTypeManager->getStorage($data['entity_type']);
}
catch (InvalidPluginDefinitionException $e) {
$message = (string) new FormattableMarkup('Invalid entity type specified using "entity_type" key when processing a job in queue @queue_id. \n @data ', [
'@queue_id' => $queueId,
'@data' => print_r($data, TRUE),
]);
$this->logger->log(RfcLogLevel::ERROR, $message);
throw new \InvalidArgumentException($message);
}
/** @var \Drupal\Core\Entity\EntityInterface $entity */
if (!isset($data['entity_id']) || !$entity = $entityStorage->load($data['entity_id'])) {
$message = (string) new FormattableMarkup('No valid entity specified by "entity_id" key when processing a job in queue @queue_id. \n @data ', [
'@queue_id' => $queueId,
'@data' => print_r($data, TRUE),
]);
$this->logger->log(RfcLogLevel::ERROR, $message);
throw new \InvalidArgumentException($message);
}
}
try {
$methodNameFragment = str_replace('_', '', ucwords($queueId, '_'));
$methodName = 'doProcess' . $methodNameFragment;
$this->{$methodName}($entity, $data);
return (string) new FormattableMarkup("Success in queue @queue for @entity_type @entity_label (@entity_id).", [
'@queue' => $queueId,
'@entity_type' => $entity->getEntityTypeId() ?: '',
'@entity_label' => $entity->label() ?: '',
'@entity_id' => $entity->id() ?: '',
]);
}
catch (\Throwable $e) {
if ($e instanceof RequestException) {
if ($response = $e->getResponse()) {
if ($response->getStatusCode(429) && $response->hasHeader('Retry-After')) {
$retryAfter = intval($response->getHeader('Retry-After')[0]);
$message = "Xero rate limit exceeded while processing $queueId queue, retry in $retryAfter seconds.";
$this->logger->log(RfcLogLevel::WARNING, $message);
self::$waitUntil = $this->time->getCurrentTime() + $retryAfter;
throw new SyncRetryException($message, $retryAfter);
}
}
}
elseif ($e instanceof XeroInvalidConfigurationException) {
$message = "Xero client unavailable when trying to process job in $queueId queue.";
$this->logger->log(RfcLogLevel::ERROR, $message);
// Wait an hour for configuration to be fixed.
$retryAfter = 3600;
self::$waitUntil = $this->time->getCurrentTime();
throw new SyncRetryException($message, $retryAfter);
}
$template = "Exception in queue @queue for @entity_type @entity_label (@entity_id). \nException message: @message \nQueue data: @data";
$params = [
'@queue' => $queueId,
'@entity_type' => $entity->getEntityTypeId() ?: '',
'@entity_label' => $entity->label() ?: '',
'@entity_id' => $entity->id() ?: '',
'@message' => $e->getMessage(),
'@data' => print_r($data, TRUE),
];
$this->logger->log(RfcLogLevel::ERROR, (string) new FormattableMarkup("$template \nException trace:\n@trace", $params + [
'@trace' => $e->getTraceAsString(),
]));
throw new SyncFailureException((string) new FormattableMarkup($template, $params));
}
}
/**
* Process an item in the xero_sync_sync queue.
*
* @param \Drupal\Core\Entity\EntityInterface $entity
* The entity to process.
* @param array $data
* The job data.
*/
protected function doProcessXeroSyncSync(EntityInterface $entity, array $data) {
$create = $data['create'] ?? TRUE;
$update = $data['update'] ?? TRUE;
$item = NULL;
if (isset($data['item_type']) && isset($data['item_guid'])) {
$item = $this->buildXeroItemWithId($data['item_type'], $data['item_guid']);
}
$this->synchronizer->syncEntityWithItem($entity, $item, $create, $update);
}
/**
* Process an item in the xero_sync_unsync queue.
*
* @param \Drupal\Core\Entity\EntityInterface $entity
* The entity to process.
* @param array $data
* The job data.
*/
protected function doProcessXeroSyncUnsync(EntityInterface $entity, array $data) {
$item = NULL;
if (isset($data['item_type']) && isset($data['item_guid'])) {
// Don't unsync an item if it's become synced with a different entity.
// Otherwise unsyncs can undo syncs inadvertently.
$fieldId = $this->entityHandler->getFieldName($entity->getEntityTypeId(), $entity->bundle());
$storage = $this->entityTypeManager->getStorage($entity->getEntityTypeId());
$query = $storage->getQuery()->accessCheck(FALSE);
$query->condition("$fieldId.guid", $data['item_guid']);
$query->condition("$fieldId.type", $data['item_type']);
$query->condition($entity->getEntityType()->getKey('id'), $entity->id(), '<>');
$synced = $query->execute();
if ($synced) {
return;
}
$item = $this->buildXeroItemWithId($data['item_type'], $data['item_guid']);
}
$this->synchronizer->unsyncEntityWithItem($entity, $item);
}
}
