diff --git a/search_api.install b/search_api.install index ae4eb4b..e163b34 100644 --- a/search_api.install +++ b/search_api.install @@ -61,17 +61,17 @@ function search_api_schema() { 'unsigned' => TRUE, 'not null' => TRUE, ), - 'server_id' => array( - 'description' => 'The ID of the server for which this task should be executed', + 'type' => array( + 'description' => 'A keyword identifying the type of task that should be executed', 'type' => 'varchar', 'length' => 50, 'not null' => TRUE, ), - 'type' => array( - 'description' => 'A keyword identifying the type of task that should be executed', + 'server_id' => array( + 'description' => 'The ID of the server for which this task should be executed', 'type' => 'varchar', 'length' => 50, - 'not null' => TRUE, + 'not null' => FALSE, ), 'index_id' => array( 'description' => 'The ID of the index to which this task pertains, if applicable for this type', diff --git a/search_api.module b/search_api.module index 38a87be..9570499 100644 --- a/search_api.module +++ b/search_api.module @@ -13,6 +13,7 @@ use Drupal\search_api\Entity\Index; use Drupal\search_api\SearchApiException; use Drupal\search_api\Plugin\search_api\datasource\ContentEntity; +use Drupal\search_api\Task\IndexTaskManager; /** * Implements hook_help(). @@ -28,10 +29,11 @@ function search_api_help($route_name) { /** * Implements hook_cron(). * - * This will first execute any pending server tasks. After that, items will - * be indexed on all enabled indexes with a non-zero cron limit. Indexing will - * run for the time set in the cron_worker_runtime config setting (defaulting to - * 15 seconds), but will at least index one batch of items on each index. + * This will first execute pending tasks (if there are any). After that, items + * will be indexed on all enabled indexes with a non-zero cron limit. Indexing + * will run for the time set in the cron_worker_runtime config setting + * (defaulting to 15 seconds), but will at least index one batch of items on + * each index. */ function search_api_cron() { // Execute pending server tasks. @@ -50,11 +52,14 @@ function search_api_cron() { // Add items to the tracking system for all indexes for which this hasn't // happened yet. - $index_task_manager = \Drupal::getContainer()->get('search_api.index_task_manager'); + $task_manager = \Drupal::getContainer()->get('search_api.task_manager'); foreach ($indexes as $index_id => $index) { - if (!$index_task_manager->isTrackingComplete($index)) { - $index_task_manager->addItemsOnce($index); - } + $conditions = array( + 'type' => IndexTaskManager::TRACK_ITEMS_TASK_TYPE, + 'index_id' => $index_id, + ); + $task_manager->executeSingleTask($conditions); + // Filter out read-only indexes here, since we want to have tracking but not // index items for them. if ($index->isReadOnly()) { diff --git a/search_api.services.yml b/search_api.services.yml index 8f9f8b2..8d67f99 100644 --- a/search_api.services.yml +++ b/search_api.services.yml @@ -36,9 +36,17 @@ services: class: Drupal\search_api\Tracker\TrackerPluginManager parent: default_plugin_manager + search_api.datasource_task_manager: + class: Drupal\search_api\Plugin\search_api\datasource\ContentEntityTaskManager + arguments: ['@search_api.task_manager', '@entity_type.manager'] + tags: + - { name: event_subscriber } + search_api.index_task_manager: class: Drupal\search_api\Task\IndexTaskManager - arguments: ['@state', '@entity_type.manager'] + arguments: ['@search_api.task_manager', '@entity_type.manager'] + tags: + - { name: event_subscriber } search_api.results_static_cache: class: Drupal\search_api\Query\ResultsCache @@ -46,6 +54,10 @@ services: search_api.server_task_manager: class: Drupal\search_api\Task\ServerTaskManager - arguments: ['@database', '@entity_type.manager'] + arguments: ['@search_api.task_manager', '@entity_type.manager'] tags: - - { name: backend_overridable } + - { name: event_subscriber } + + search_api.task_manager: + class: Drupal\search_api\Task\TaskManager + arguments: ['@entity_type.manager', '@event_dispatcher', '@string_translation'] diff --git a/src/Entity/Server.php b/src/Entity/Server.php index 6bcb619..6633ab2 100644 --- a/src/Entity/Server.php +++ b/src/Entity/Server.php @@ -207,7 +207,7 @@ public function addIndex(IndexInterface $index) { // When freshly adding an index to a server, it doesn't make any sense to // execute possible other tasks for that server/index combination. // (removeIndex() is implicit when adding an index which was already added.) - $server_task_manager->delete(NULL, $this, $index); + $server_task_manager->delete($this, $index); try { if ($server_task_manager->execute($this)) { @@ -221,8 +221,11 @@ public function addIndex(IndexInterface $index) { '%index' => $index->label(), ); watchdog_exception('search_api', $e, '%type while adding index %index to server %server: @message in %function (line %line of %file).', $vars); - $server_task_manager->add($this, __FUNCTION__, $index); } + + $task_manager = \Drupal::getContainer() + ->get('search_api.task_manager'); + $task_manager->addTask(__FUNCTION__, $this, $index); } /** @@ -243,7 +246,10 @@ public function updateIndex(IndexInterface $index) { ); watchdog_exception('search_api', $e, '%type while updating the fields of index %index on server %server: @message in %function (line %line of %file).', $vars); } - $server_task_manager->add($this, __FUNCTION__, $index, isset($index->original) ? $index->original : NULL); + + $task_manager = \Drupal::getContainer() + ->get('search_api.task_manager'); + $task_manager->addTask(__FUNCTION__, $this, $index, isset($index->original) ? $index->original : NULL); } /** @@ -253,7 +259,7 @@ public function removeIndex($index) { $server_task_manager = \Drupal::getContainer()->get('search_api.server_task_manager'); // When removing an index from a server, it doesn't make any sense anymore // to delete items from it, or react to other changes. - $server_task_manager->delete(NULL, $this, $index); + $server_task_manager->delete($this, $index); try { if ($server_task_manager->execute($this)) { @@ -267,8 +273,16 @@ public function removeIndex($index) { '%index' => is_object($index) ? $index->label() : $index, ); watchdog_exception('search_api', $e, '%type while removing index %index from server %server: @message in %function (line %line of %file).', $vars); - $server_task_manager->add($this, __FUNCTION__, $index); } + + $task_manager = \Drupal::getContainer() + ->get('search_api.task_manager'); + $data = NULL; + if (!is_object($index)) { + $data = $index; + $index = NULL; + } + $task_manager->addTask(__FUNCTION__, $this, $index, $data); } /** @@ -308,7 +322,10 @@ public function deleteItems(IndexInterface $index, array $item_ids) { ); watchdog_exception('search_api', $e, '%type while deleting items from server %server: @message in %function (line %line of %file).', $vars); } - $server_task_manager->add($this, __FUNCTION__, $index, $item_ids); + + $task_manager = \Drupal::getContainer() + ->get('search_api.task_manager'); + $task_manager->addTask(__FUNCTION__, $this, $index, $item_ids); } /** @@ -331,7 +348,7 @@ public function deleteAllIndexItems(IndexInterface $index) { 'deleteItems', 'deleteAllIndexItems', ); - $server_task_manager->delete(NULL, $this, $index, $types); + $server_task_manager->delete($this, $index, $types); try { if ($server_task_manager->execute($this)) { @@ -346,7 +363,10 @@ public function deleteAllIndexItems(IndexInterface $index) { ); watchdog_exception('search_api', $e, '%type while deleting items of index %index from server %server: @message in %function (line %line of %file).', $vars); } - $server_task_manager->add($this, __FUNCTION__, $index); + + $task_manager = \Drupal::getContainer() + ->get('search_api.task_manager'); + $task_manager->addTask(__FUNCTION__, $this, $index); } /** @@ -383,7 +403,7 @@ public function deleteAllItems() { ); \Drupal::getContainer() ->get('search_api.server_task_manager') - ->delete(NULL, $this, NULL, $types); + ->delete($this, NULL, $types); } /** @@ -454,7 +474,7 @@ public static function preDelete(EntityStorageInterface $storage, array $entitie if ($server->hasValidBackend()) { $server->getBackend()->preDelete(); } - \Drupal::getContainer()->get('search_api.server_task_manager')->delete(NULL, $server); + \Drupal::getContainer()->get('search_api.server_task_manager')->delete($server); } } diff --git a/src/Entity/Task.php b/src/Entity/Task.php new file mode 100644 index 0000000..9982fc9 --- /dev/null +++ b/src/Entity/Task.php @@ -0,0 +1,178 @@ +entityTypeManager ?: \Drupal::entityTypeManager(); + } + + /** + * {@inheritdoc} + */ + public function setEntityTypeManager(EntityTypeManagerInterface $entityTypeManager) { + $this->entityTypeManager = $entityTypeManager; + return $this; + } + + /** + * {@inheritdoc} + */ + public function getType() { + return $this->get('type')[0]->value; + } + + /** + * {@inheritdoc} + */ + public function getServerId() { + $field = $this->get('server_id')[0]; + return $field ? $field->value : NULL; + } + + /** + * {@inheritdoc} + */ + public function getServer() { + if ($this->getServerId() && !isset($this->serverInstance)) { + $this->serverInstance = $this->getEntityTypeManager() + ->getStorage('search_api_server') + ->load($this->getServerId()); + if (!$this->serverInstance) { + $args['%server'] = $this->getServerId(); + throw new SearchApiException(new FormattableMarkup('Could not load server with ID %server.', $args)); + } + } + + return $this->serverInstance; + } + + /** + * {@inheritdoc} + */ + public function getIndexId() { + $field = $this->get('index_id')[0]; + return $field ? $field->value : NULL; + } + + /** + * {@inheritdoc} + */ + public function getIndex() { + if ($this->getIndexId() && !isset($this->indexInstance)) { + $this->indexInstance = $this->getEntityTypeManager() + ->getStorage('search_api_index') + ->load($this->getIndexId()); + if (!$this->indexInstance) { + $args['%index'] = $this->getIndexId(); + throw new SearchApiException(new FormattableMarkup('Could not load index with ID %index.', $args)); + } + } + + return $this->indexInstance; + } + + /** + * {@inheritdoc} + */ + public function getData() { + if (!isset($this->unserializedData)) { + $data = $this->get('data')[0]; + if ($data) { + $this->unserializedData = unserialize($data->value); + } + } + + return $this->unserializedData; + } + + /** + * {@inheritdoc} + */ + public static function baseFieldDefinitions(EntityTypeInterface $entity_type) { + /** @var \Drupal\Core\Field\BaseFieldDefinition[] $fields */ + $fields = parent::baseFieldDefinitions($entity_type); + + $fields['type'] = BaseFieldDefinition::create('string') + ->setLabel(t('Task type')) + ->setSetting('max_length', 50) + ->setRequired(TRUE) + ->setReadOnly(TRUE); + + $fields['server_id'] = BaseFieldDefinition::create('string') + ->setLabel(t('Server ID')) + ->setSetting('max_length', 50) + ->setReadOnly(TRUE); + + $fields['index_id'] = BaseFieldDefinition::create('string') + ->setLabel(t('Index ID')) + ->setSetting('max_length', 50) + ->setReadOnly(TRUE); + + $fields['data'] = BaseFieldDefinition::create('string_long') + ->setLabel(t('Task data')) + ->setReadOnly(TRUE); + + return $fields; + } + +} diff --git a/src/Plugin/search_api/datasource/ContentEntity.php b/src/Plugin/search_api/datasource/ContentEntity.php index ccf1c4b..abb61a6 100644 --- a/src/Plugin/search_api/datasource/ContentEntity.php +++ b/src/Plugin/search_api/datasource/ContentEntity.php @@ -24,6 +24,7 @@ use Drupal\search_api\Entity\Index; use Drupal\search_api\SearchApiException; use Drupal\search_api\IndexInterface; +use Drupal\search_api\Task\TaskManager; use Drupal\search_api\Utility; use Symfony\Component\DependencyInjection\ContainerInterface; @@ -35,7 +36,7 @@ * deriver = "Drupal\search_api\Plugin\search_api\datasource\ContentEntityDeriver" * ) */ -class ContentEntity extends DatasourcePluginBase { +class ContentEntity extends DatasourcePluginBase implements EntityDatasourceInterface { /** * The entity type manager. @@ -110,33 +111,14 @@ public static function create(ContainerInterface $container, array $configuratio /** @var static $datasource */ $datasource = parent::create($container, $configuration, $plugin_id, $plugin_definition); - /** @var $entity_type_manager \Drupal\Core\Entity\EntityTypeManagerInterface */ - $entity_type_manager = $container->get('entity_type.manager'); - $datasource->setEntityTypeManager($entity_type_manager); - - /** @var $entity_field_manager \Drupal\Core\Entity\EntityFieldManagerInterface */ - $entity_field_manager = $container->get('entity_field.manager'); - $datasource->setEntityFieldManager($entity_field_manager); - - /** @var $entity_display_repo \Drupal\Core\Entity\EntityDisplayRepositoryInterface */ - $entity_display_repo = $container->get('entity_display.repository'); - $datasource->setEntityDisplayRepository($entity_display_repo); - - /** @var $entity_type_bundle_info \Drupal\Core\Entity\EntityTypeBundleInfoInterface */ - $entity_type_bundle_info = $container->get('entity_type.bundle.info'); - $datasource->setEntityTypeBundleInfo($entity_type_bundle_info); - - /** @var \Drupal\Core\TypedData\TypedDataManager $typed_data_manager */ - $typed_data_manager = $container->get('typed_data_manager'); - $datasource->setTypedDataManager($typed_data_manager); - - /** @var $config_factory \Drupal\Core\Config\ConfigFactoryInterface */ - $config_factory = $container->get('config.factory'); - $datasource->setConfigFactory($config_factory); - - /** @var $language_manager \Drupal\Core\Language\LanguageManagerInterface */ - $language_manager = $container->get('language_manager'); - $datasource->setLanguageManager($language_manager); + $datasource->setEntityTypeManager($container->get('entity_type.manager')); + $datasource->setEntityFieldManager($container->get('entity_field.manager')); + $datasource->setEntityDisplayRepository($container->get('entity_display.repository')); + $datasource->setEntityTypeBundleInfo($container->get('entity_type.bundle.info')); + $datasource->setTypedDataManager($container->get('typed_data_manager')); + $datasource->setConfigFactory($container->get('config.factory')); + $datasource->setLanguageManager($container->get('language_manager')); + $datasource->setTaskManager($container->get('search_api.task_manager')); return $datasource; } @@ -333,6 +315,36 @@ public function setLanguageManager(LanguageManagerInterface $language_manager) { } /** + * The task manager. + * + * @var \Drupal\search_api\Task\TaskManager|null + */ + protected $taskManager; + + /** + * Retrieves the task manager. + * + * @return \Drupal\search_api\Task\TaskManager + * The task manager. + */ + public function getTaskManager() { + return $this->taskManager ?: \Drupal::service('search_api.task_manager'); + } + + /** + * Sets the task manager. + * + * @param \Drupal\search_api\Task\TaskManager $task_manager + * The new task manager. + * + * @return $this + */ + public function setTaskManager(TaskManager $task_manager) { + $this->taskManager = $task_manager; + return $this; + } + + /** * {@inheritdoc} */ public function getPropertyDefinitions() { @@ -350,17 +362,7 @@ public function getPropertyDefinitions() { * {@inheritdoc} */ public function loadMultiple(array $ids) { - $allowed_languages = $all_languages = $this->getLanguageManager()->getLanguages(); - - if ($this->isTranslatable()) { - $selected_languages = array_flip($this->configuration['languages']['selected']); - if ($this->configuration['languages']['default']) { - $allowed_languages = array_diff_key($all_languages, $selected_languages); - } - else { - $allowed_languages = array_intersect_key($all_languages, $selected_languages); - } - } + $allowed_languages = $this->getLanguages(); // Always allow items with undefined language. (Can be the case when // entities are created programmatically.) $allowed_languages[LanguageInterface::LANGCODE_NOT_SPECIFIED] = TRUE; @@ -417,38 +419,67 @@ public function setConfiguration(array $new_config) { // We'd now check the items of which bundles need to be added to or removed // from tracking for this index. However, if the index is not tracking at // all (i.e., is disabled) currently, we can skip all that. - if (!$this->index->status() || !$this->index->hasValidTracker()) { + if (!$this->index->status()) { return; } - if ($this->hasBundles() && $old_config != $new_config) { - $old_bundles = array_flip($old_config['bundles']['selected']); - $new_bundles = array_flip($new_config['bundles']['selected']); - - // First, check if the "default" setting changed and invert the set - // bundles for the old config, so the following comparison makes sense. - if ($old_config['bundles']['default'] != $new_config['bundles']['default']) { - $old_bundles = array_diff_key($this->getBundles(), $old_bundles); + if ($old_config != $new_config) { + // Bundles and languages share the same structure, so changes can be + // processed in a unified way. + $tasks = array(); + $insert_task = ContentEntityTaskManager::INSERT_ITEMS_TASK_TYPE; + $delete_task = ContentEntityTaskManager::DELETE_ITEMS_TASK_TYPE; + $settings = array(); + if ($this->hasBundles()) { + $settings['bundles'] = $this->getBundles(); } + if ($this->isTranslatable()) { + $settings['languages'] = $this->getAllLanguages(); + } + + // Determine which bundles/languages have been newly selected or + // deselected and then assign them to the appropriate actions depending + // on the current "default" setting. + foreach ($settings as $setting => $all) { + $old_selected = array_flip($old_config[$setting]['selected']); + $new_selected = array_flip($new_config[$setting]['selected']); + + // First, check if the "default" setting changed and invert the checked + // items for the old config, so the following comparison makes sense. + if ($old_config[$setting]['default'] != $new_config[$setting]['default']) { + $old_selected = array_diff_key($all, $old_selected); + } - // Now, go through all the bundles and start/stop tracking for them - // accordingly. - $newly_selected = array_diff_key($new_bundles, $old_bundles); - $newly_unselected = array_diff_key($old_bundles, $new_bundles); - if ($new_config['bundles']['default']) { - $actions['trackItemsDeleted'] = $newly_selected; - $actions['trackItemsInserted'] = $newly_unselected; + $newly_selected = array_diff_key($new_selected, $old_selected); + $newly_unselected = array_diff_key($old_selected, $new_selected); + if ($new_config[$setting]['default']) { + $tasks[$insert_task][$setting] = $newly_unselected ?: NULL; + $tasks[$delete_task][$setting] = $newly_selected ?: NULL; + } + else { + $tasks[$insert_task][$setting] = $newly_selected ?: NULL; + $tasks[$delete_task][$setting] = $newly_unselected ?: NULL; + } } - else { - $actions['trackItemsDeleted'] = $newly_unselected; - $actions['trackItemsInserted'] = $newly_selected; + + // This will keep only those tasks where at least one of "bundles" or + // "languages" is non-empty. + $tasks = array_filter($tasks, 'array_filter'); + foreach ($tasks as $task => $data) { + $data += array( + 'datasource' => $this->getPluginId(), + 'page' => 0, + ); + $this->getTaskManager()->addTask($task, NULL, $this->index, $data); } - $actions = array_filter($actions); - foreach ($actions as $action => $bundles) { - if ($entity_ids = $this->getBundleItemIds(array_keys($bundles))) { - $this->getIndex()->$action($this->getPluginId(), $entity_ids); - } + // If we added any new tasks, set a batch for them. (If we aren't in a + // form submission, this will just be ignored.) + if ($tasks) { + $this->getTaskManager()->setTasksBatch(array( + 'index_id' => $this->index->id(), + 'type' => array_keys($tasks), + )); } } } @@ -557,7 +588,7 @@ protected function getEntityBundleOptions() { */ protected function getTranslationOptions() { $options = array(); - foreach ($this->getLanguageManager()->getLanguages() as $language) { + foreach ($this->getAllLanguages() as $language) { $options[$language->getId()] = $language->getName(); } @@ -642,7 +673,7 @@ public function getItemUrl(ComplexDataInterface $item) { * {@inheritdoc} */ public function getItemIds($page = NULL) { - return $this->getBundleItemIds(NULL, $page); + return $this->getPartialItemIds($page); } /** @@ -718,48 +749,70 @@ protected function getEntityBundles() { } /** - * Retrieves all item IDs of entities of the specified bundles. - * - * @param string[]|null $bundles - * (optional) The bundles for which all item IDs should be returned; or NULL - * to retrieve IDs from all enabled bundles in this datasource. - * @param int|null $page - * The zero-based page of IDs to retrieve, for the paging mechanism - * implemented by this datasource; or NULL to retrieve all items at once. - * - * @return string[] - * An array of all item IDs of these bundles. + * {@inheritdoc} */ - protected function getBundleItemIds(array $bundles = NULL, $page = NULL) { - // If NULL was passed, use all enabled bundles. - if (!isset($bundles)) { - $bundles = array_keys($this->getBundles()); - } - + public function getPartialItemIds($page = NULL, array $bundles = NULL, array $languages = NULL) { $select = \Drupal::entityQuery($this->getEntityTypeId()); - // If there are bundles to filter on, and they don't include all available - // bundles, add the appropriate condition. - if ($bundles && $this->hasBundles()) { - if (count($bundles) != count($this->getEntityBundles())) { - $select->condition($this->getEntityType()->getKey('bundle'), $bundles, 'IN'); + + // We want to determine all entities of either one of the given bundles OR + // one of the given languages. That means we can't just filter for $bundles + // if $languages is given, but instead on all bundles we might want to + // include and later sort out those for which we want only the translations + // in $languages and those (matching $bundles) where we want all (enabled) + // translations. + if ($this->hasBundles()) { + $bundle_property = $this->getEntityType()->getKey('bundle'); + if ($bundles && !$languages) { + $select->condition($bundle_property, $bundles, 'IN'); + } + else { + $enabled_bundles = array_keys($this->getBundles()); + // Since this is also called for removed bundles/languages, + // $enabled_bundles might not include $bundles. + if ($bundles) { + $enabled_bundles = array_unique(array_merge($bundles, $enabled_bundles)); + } + if (count($enabled_bundles) < count($this->getEntityBundles())) { + $select->condition($bundle_property, $enabled_bundles, 'IN'); + } } } + if (isset($page)) { $page_size = $this->getConfigValue('tracking_page_size'); $select->range($page * $page_size, $page_size); } + $entity_ids = $select->execute(); if (!$entity_ids) { return NULL; } - // For all the loaded entities, compute all their item IDs (one for each - // translation). + // For all loaded entities, compute all their item IDs (one for each + // translation we want to include). For those matching the given bundles (if + // any), we want to include translations for all enabled languages. For all + // other entities, we just want to include the translations for the + // languages passed to the method (if any). $item_ids = array(); + $enabled_languages = array_keys($this->getLanguages()); + // As above for bundles, $enabled_languages might not include $languages. + if ($languages) { + $enabled_languages = array_unique(array_merge($languages, $enabled_languages)); + } + // Also, we want to always include entities with unknown language. + $enabled_languages[] = LanguageInterface::LANGCODE_NOT_SPECIFIED; + /** @var \Drupal\Core\Entity\ContentEntityInterface $entity */ foreach ($this->getEntityStorage()->loadMultiple($entity_ids) as $entity_id => $entity) { - foreach (array_keys($entity->getTranslationLanguages()) as $langcode) { + $translations = array_keys($entity->getTranslationLanguages()); + if (!isset($bundles) || in_array($entity->bundle(), $bundles)) { + $translations = array_intersect($translations, $enabled_languages); + } + else { + $translations = array_intersect($translations, $languages); + } + foreach ($translations as $langcode) { $item_ids[] = "$entity_id:$langcode"; } } @@ -793,6 +846,39 @@ public function getBundles() { } /** + * Retrieves the enabled languages. + * + * @return \Drupal\Core\Language\LanguageInterface[] + * All languages that are enabled for this datasource, keyed by language + * code. + */ + protected function getLanguages() { + $all_languages = $this->getAllLanguages(); + + if ($this->isTranslatable()) { + $selected_languages = array_flip($this->configuration['languages']['selected']); + if ($this->configuration['languages']['default']) { + return array_diff_key($all_languages, $selected_languages); + } + else { + return array_intersect_key($all_languages, $selected_languages); + } + } + + return $all_languages; + } + + /** + * Retrieves all available languages. + * + * @return \Drupal\Core\Language\LanguageInterface[] + * All available languages on this site, keyed by language code. + */ + protected function getAllLanguages() { + return $this->getLanguageManager()->getLanguages(); + } + + /** * {@inheritdoc} */ public function getViewModes($bundle = NULL) { @@ -943,18 +1029,7 @@ protected function getPropertyPathDependencies($property_path, array $properties } /** - * Returns an array of config entity dependencies. - * - * @param string $entity_type_id - * The entity type to which the fields are attached. - * @param string[] $fields - * An array of property paths of fields from this entity type. - * @param string[] $all_fields - * An array of property paths of all the fields from this datasource. - * - * @return string[] - * An array keyed by the IDs of entities on which this datasource depends. - * The values are containing list of Search API fields. + * {@inheritdoc} */ public function getFieldDependenciesForEntityType($entity_type_id, array $fields, array $all_fields) { $field_dependencies = array(); @@ -1010,14 +1085,7 @@ public function getFieldDependenciesForEntityType($entity_type_id, array $fields } /** - * Retrieves all indexes that are configured to index the given entity. - * - * @param \Drupal\Core\Entity\ContentEntityInterface $entity - * The entity for which to check. - * - * @return \Drupal\search_api\IndexInterface[] - * All indexes that are configured to index the given entity (using this - * datasource class). + * {@inheritdoc} */ public static function getIndexesForEntity(ContentEntityInterface $entity) { $entity_type = $entity->getEntityTypeId(); diff --git a/src/Plugin/search_api/datasource/ContentEntityTaskManager.php b/src/Plugin/search_api/datasource/ContentEntityTaskManager.php new file mode 100644 index 0000000..0970b9a --- /dev/null +++ b/src/Plugin/search_api/datasource/ContentEntityTaskManager.php @@ -0,0 +1,116 @@ +taskManager = $task_manager; + $this->entityTypeManager = $entity_type_manager; + } + + /** + * {@inheritdoc} + */ + public static function getSubscribedEvents() { + $events['search_api.task.' . static::INSERT_ITEMS_TASK_TYPE][] = array('processEvent'); + $events['search_api.task.' . static::DELETE_ITEMS_TASK_TYPE][] = array('processEvent'); + + return $events; + } + + /** + * Processes a datasource tracking event. + * + * @param \Drupal\search_api\Task\TaskEvent $event + * The task event. + * @param string $event_name + * The name of the event. + */ + public function processEvent(TaskEvent $event, $event_name) { + $event->stopPropagation(); + + // The complete event name prefix in front of the method name is 45 + // characters long: "search_api.task.search_api.entity_datasource.". + $method = substr($event_name, 45); + + $task = $event->getTask(); + $index = $task->getIndex(); + $data = $task->getData(); + + if (!$index->hasValidTracker()) { + $args['%index'] = $index->label(); + $message = new FormattableMarkup('Index %index does not have a valid tracker set.', $args); + $event->setException(new SearchApiException($message)); + return; + } + + $datasource_id = $data['datasource']; + $reschedule = FALSE; + if ($index->isValidDatasource($datasource_id)) { + $datasource = $index->getDatasource($datasource_id); + if ($datasource instanceof EntityDatasourceInterface) { + $raw_ids = $datasource->getPartialItemIds($data['page'], $data['bundles'], $data['languages']); + if ($raw_ids !== NULL) { + $reschedule = TRUE; + if ($raw_ids) { + $item_ids = array(); + foreach ($raw_ids as $raw_id) { + $item_ids[] = Utility::createCombinedId($datasource_id, $raw_id); + } + $index->getTrackerInstance()->$method($item_ids); + } + } + } + } + + if ($reschedule) { + ++$data['page']; + $this->taskManager->addTask($method, NULL, $index, $data); + } + } + +} diff --git a/src/Plugin/search_api/datasource/EntityDatasourceInterface.php b/src/Plugin/search_api/datasource/EntityDatasourceInterface.php new file mode 100644 index 0000000..1101ca7 --- /dev/null +++ b/src/Plugin/search_api/datasource/EntityDatasourceInterface.php @@ -0,0 +1,60 @@ +state = $state; + public function __construct(TaskManagerInterface $task_manager, EntityTypeManagerInterface $entity_type_manager) { + $this->taskManager = $task_manager; $this->entityTypeManager = $entity_type_manager; } /** * {@inheritdoc} */ - public function startTracking(IndexInterface $index, array $datasource_ids = NULL) { - $index_state = $this->getIndexState($index); - if (empty($index_state['status'])) { - if (!isset($datasource_ids)) { - $datasource_ids = $index->getDatasourceIds(); - } - elseif (!$datasource_ids) { - // Calling with an empty array as $datasource_ids could otherwise break - // this, so we have to check for this situation. - return; - } - $index_state = array( - 'status' => 1, - 'pages' => array_fill_keys($datasource_ids, 0), - ); - $this->state->set($this->getIndexStateKey($index), $index_state); - } + public static function getSubscribedEvents() { + $events['search_api.task.' . static::TRACK_ITEMS_TASK_TYPE][] = array('trackItems'); + + return $events; } /** - * {@inheritdoc} + * Tracks items according to the given event. + * + * @param \Drupal\search_api\Task\TaskEvent $event + * The task event. */ - public function addItemsOnce(IndexInterface $index) { - $index_state = $this->getIndexState($index); - if (!($index_state['status'] && $index_state['pages'])) { - return NULL; - } + public function trackItems(TaskEvent $event) { + $event->stopPropagation(); + + $task = $event->getTask(); + $index = $task->getIndex(); if (!$index->hasValidTracker()) { - return 0; + $args['%index'] = $index->label(); + $message = new FormattableMarkup('Index %index does not have a valid tracker set.', $args); + $event->setException(new SearchApiException($message)); + return; } - // Use this method to automatically circle through the datasources, adding - // items from each of them in turn. - $page = reset($index_state['pages']); - $datasource_id = key($index_state['pages']); - unset($index_state['pages'][$datasource_id]); + $data = $task->getData(); + $datasource_id = $data['datasource']; - $added = 0; + $reschedule = FALSE; if ($index->isValidDatasource($datasource_id)) { - $raw_ids = $index->getDatasource($datasource_id)->getItemIds($page); + $raw_ids = $index->getDatasource($datasource_id)->getItemIds($data['page']); if ($raw_ids !== NULL) { - $index_state['pages'][$datasource_id] = ++$page; + $reschedule = TRUE; if ($raw_ids) { $item_ids = array(); foreach ($raw_ids as $raw_id) { $item_ids[] = Utility::createCombinedId($datasource_id, $raw_id); } - $added = count($item_ids); $index->getTrackerInstance()->trackItemsInserted($item_ids); } } } - if (empty($index_state['pages'])) { - $this->state->delete($this->getIndexStateKey($index)); - return NULL; + if ($reschedule) { + ++$data['page']; + $this->taskManager->addTask(static::TRACK_ITEMS_TASK_TYPE, NULL, $index, $data); } - - $this->state->set($this->getIndexStateKey($index), $index_state); - return $added; } /** * {@inheritdoc} */ - public function addItemsBatch(IndexInterface $index) { - $index_state = $this->getIndexState($index); - if (!empty($index_state['status'])) { - $batch_definition = array( - 'operations' => array( - array(array($this, 'processBatch'), array($index->id())), - ), - 'finished' => array($this, 'finishBatch'), + public function startTracking(IndexInterface $index, array $datasource_ids = NULL) { + if (!isset($datasource_ids)) { + $datasource_ids = $index->getDatasourceIds(); + } + + foreach ($datasource_ids as $datasource_id) { + $data = array( + 'datasource' => $datasource_id, + 'page' => 0, ); - // Schedule the batch. - batch_set($batch_definition); + $this->taskManager->addTask(static::TRACK_ITEMS_TASK_TYPE, NULL, $index, $data); } } /** - * Adds one page of items to the tracker as part of a batch operation. + * Gets a set of conditions for finding the tracking tasks of the given index. * - * @param string $index_id - * The ID of the index on which items should be indexed. - * @param array $context - * The current batch context, as defined in the @link batch Batch operations - * @endlink documentation. + * @param \Drupal\search_api\IndexInterface $index + * The index for which to retrieve tasks. + * + * @return array + * An array of conditions to pass to the Search API task manager. */ - public function processBatch($index_id, array &$context) { - /** @var \Drupal\search_api\IndexInterface $index */ - $index = $this->entityTypeManager->getStorage('search_api_index')->load($index_id); - $added = $this->addItemsOnce($index); - if (!isset($context['results']['added'])) { - $context['results']['added'] = 0; - } - if ($added === NULL) { - $context['finished'] = 1; - } - else { - $context['finished'] = 0; - $context['results']['added'] += $added; - $args['%index'] = $index->label(); - $context['message'] = $this->formatPlural($context['results']['added'], 'Tracked 1 item for the %index index.', 'Tracked @count items for the %index index.', $args); - } + protected function getTaskConditions(IndexInterface $index) { + return array( + 'type' => static::TRACK_ITEMS_TASK_TYPE, + 'index_id' => $index->id(), + ); } /** - * Finishes a "start tracking" batch. - * - * @param bool $success - * Indicates whether the batch process was successful. - * @param array $results - * Results information passed from the processing callback. + * {@inheritdoc} */ - public function finishBatch($success, $results) { - // Check if the batch job was successful. - if ($success) { - // Display the number of items tracked. - $indexed_message = $this->formatPlural($results['added'], 'Successfully tracked 1 item for this index.', 'Successfully tracked @count items for this index.'); - drupal_set_message($indexed_message); - } - else { - // Notify the user about the batch job failure. - drupal_set_message($this->t('An error occurred while trying to start tracking. Check the logs for details.'), 'error'); - } + public function addItemsOnce(IndexInterface $index) { + return !$this->taskManager->executeSingleTask($this->getTaskConditions($index)); + } + + /** + * {@inheritdoc} + */ + public function addItemsBatch(IndexInterface $index) { + $this->taskManager->setTasksBatch($this->getTaskConditions($index)); } /** * {@inheritdoc} */ public function addItemsAll(IndexInterface $index) { - do { - $return = $this->addItemsOnce($index); - } while ($return !== NULL); + $this->taskManager->executeAllTasks($this->getTaskConditions($index)); } /** @@ -188,7 +161,7 @@ public function addItemsAll(IndexInterface $index) { public function stopTracking(IndexInterface $index, array $datasource_ids = NULL) { $valid_tracker = $index->hasValidTracker(); if (!isset($datasource_ids)) { - $this->state->delete($this->getIndexStateKey($index)); + $this->taskManager->deleteTasks($this->getTaskConditions($index)); if ($valid_tracker) { $index->getTrackerInstance()->trackAllItemsDeleted(); } @@ -200,23 +173,12 @@ public function stopTracking(IndexInterface $index, array $datasource_ids = NULL return; } - // If no state is saved, this will return NULL, making the following unset() - // statements no-ops. - $index_state = $this->getIndexState($index, FALSE); - foreach ($datasource_ids as $datasource_id) { - unset($index_state['pages'][$datasource_id]); - if ($valid_tracker) { - $index->getTrackerInstance()->trackAllItemsDeleted($datasource_id); - } - } - - // If we had an index state saved, update it now. - if (isset($index_state)) { - if (empty($index_state['pages'])) { - $this->state->delete($this->getIndexStateKey($index)); - } - else { - $this->state->set($this->getIndexStateKey($index), $index_state); + $datasource_ids = array_flip($datasource_ids); + $tasks = $this->taskManager->loadTasks($this->getTaskConditions($index)); + foreach ($tasks as $task_id => $task) { + $data = $task->getData(); + if (isset($datasource_ids[$data['datasource']])) { + $this->taskManager->deleteTask($task_id); } } } @@ -225,63 +187,7 @@ public function stopTracking(IndexInterface $index, array $datasource_ids = NULL * {@inheritdoc} */ public function isTrackingComplete(IndexInterface $index) { - $index_state = $this->getIndexState($index); - return !$index_state['status']; - } - - /** - * Retrieves the current state for a specific index. - * - * @param \Drupal\search_api\IndexInterface $index - * The index whose state should be retrieved. - * @param bool $provide_default - * Whether to provide a default state array if there was no state saved. - * - * @return array|null - * An associative array with information about the state of the given index: - * - status: 0 if no tracking operation is currently in progress, 1 if there - * is. - * - page: If status is 1, the next page of items to process for tracking. - * Or NULL if there was no saved state and $provide_default is FALSE. - */ - protected function getIndexState(IndexInterface $index, $provide_default = TRUE) { - $default = array( - 'status' => 0, - 'pages' => array(), - ); - return $this->state->get($this->getIndexStateKey($index), $provide_default ? $default : NULL); - } - - /** - * Retrieves the key of the index's state. - * - * @param \Drupal\search_api\IndexInterface $index - * The index. - * - * @return string - * The key which identifies the index's state in the site state. - */ - protected function getIndexStateKey(IndexInterface $index) { - return 'search_api.index.' . $index->id() . '.tasks'; - } - - /** - * Implements the magic __sleep() method. - * - * Prevents the any properties from being serialized. - */ - public function __sleep() { - return array(); - } - - /** - * Implements the magic __wakeup() method. - * - * Reloads the services into this object's properties. - */ - public function __wakeup() { - $this->state = \Drupal::state(); - $this->entityTypeManager = \Drupal::entityTypeManager(); + return !$this->taskManager->getTasksCount($this->getTaskConditions($index)); } } diff --git a/src/Task/IndexTaskManagerInterface.php b/src/Task/IndexTaskManagerInterface.php index 1e5b393..410fb81 100644 --- a/src/Task/IndexTaskManagerInterface.php +++ b/src/Task/IndexTaskManagerInterface.php @@ -26,10 +26,11 @@ public function startTracking(IndexInterface $index, array $datasource_ids = NUL * @param \Drupal\search_api\IndexInterface $index * The search index. * - * @return int|null - * The number of items tracked. Or NULL if no items were added and tracking - * for this index has been completed. (Note that 0 can also be returned, - * which does not mean that tracking has been completed for the index.) + * @return bool + * TRUE if tracking for this index has been finished, FALSE otherwise. + * + * @throws \Drupal\search_api\SearchApiException + * Thrown if any error occurred while tracking items. */ public function addItemsOnce(IndexInterface $index); @@ -50,10 +51,8 @@ public function addItemsBatch(IndexInterface $index); * @param \Drupal\search_api\IndexInterface $index * The search index. * - * @return int|null - * The number of items tracked. Or NULL if no items were added and tracking - * for this index has been completed. (Note that 0 can also be returned, - * which does not mean that tracking has been completed for the index.) + * @throws \Drupal\search_api\SearchApiException + * Thrown if any error occurred while tracking items. */ public function addItemsAll(IndexInterface $index); diff --git a/src/Task/ServerTaskManager.php b/src/Task/ServerTaskManager.php index 02558e2..3ecba15 100644 --- a/src/Task/ServerTaskManager.php +++ b/src/Task/ServerTaskManager.php @@ -2,29 +2,27 @@ namespace Drupal\search_api\Task; -use Drupal\Core\Database\Connection; -use Drupal\Core\Database\Query\SelectInterface; -use Drupal\Core\DependencyInjection\DependencySerializationTrait; +use Drupal\Component\Render\FormattableMarkup; use Drupal\Core\Entity\EntityTypeManagerInterface; use Drupal\Core\StringTranslation\StringTranslationTrait; -use Drupal\search_api\SearchApiException; use Drupal\search_api\IndexInterface; +use Drupal\search_api\SearchApiException; use Drupal\search_api\ServerInterface; +use Symfony\Component\EventDispatcher\EventSubscriberInterface; /** * Provides a service for managing pending server tasks. */ -class ServerTaskManager implements ServerTaskManagerInterface { +class ServerTaskManager implements ServerTaskManagerInterface, EventSubscriberInterface { use StringTranslationTrait; - use DependencySerializationTrait; /** - * The database connection. + * The Search API task manager. * - * @var \Drupal\Core\Database\Connection + * @var \Drupal\search_api\Task\TaskManagerInterface */ - protected $database; + protected $taskManager; /** * The entity type manager. @@ -36,137 +34,27 @@ class ServerTaskManager implements ServerTaskManagerInterface { /** * Constructs a ServerTaskManager object. * - * @param \Drupal\Core\Database\Connection $database - * The database connection. + * @param \Drupal\search_api\Task\TaskManagerInterface $task_manager + * The Search API task manager. * @param \Drupal\Core\Entity\EntityTypeManagerInterface $entity_type_manager * The entity type manager. */ - public function __construct(Connection $database, EntityTypeManagerInterface $entity_type_manager) { - $this->database = $database; + public function __construct(TaskManagerInterface $task_manager, EntityTypeManagerInterface $entity_type_manager) { + $this->taskManager = $task_manager; $this->entityTypeManager = $entity_type_manager; } /** * {@inheritdoc} */ - public function getCount(ServerInterface $server = NULL) { - return $this->createQuery($server) - ->range() - ->countQuery() - ->execute() - ->fetchField(); - } - - /** - * {@inheritdoc} - */ - public function execute(ServerInterface $server = NULL) { - if ($server && !$server->status()) { - return FALSE; - } - - $select = $this->createQuery($server); - - // Execute the SELECT query and remember the limit used. (Weirdly, the - // query object doesn't seem to have a method for retrieving the range after - // setting it.) - $tasks = $select->execute(); - $fetch_count = $select->getMetaData('search_api_server_tasks_limit'); - - // After all possible modifications to the query have been made, we can now - // also determine the total task count we would have had. Unfortunately, - // since we had to set the range before (which countQuery() doesn't - // automatically remove for some reason), this isn't completely - // straight-forward. - $count_query = clone $select; - $total_count = $count_query->range()->countQuery()->execute()->fetchField(); - - $executed_tasks = array(); - $failing_servers = array(); - foreach ($tasks as $task) { - if (isset($failing_servers[$task->server_id])) { - continue; - } - if (!$server || $server->id() != $task->server_id) { - $server = $this->loadServer($task->server_id); - if (!$server) { - $failing_servers[$task->server_id] = TRUE; - continue; - } - } - if (!$server->status()) { - continue; - } - $index = NULL; - if ($task->index_id) { - $index = $this->loadIndex($task->index_id); - } - try { - if ($this->executeTask($task, $server, $index)) { - $executed_tasks[] = $task->id; - } - } - catch (SearchApiException $e) { - // If a task fails, we don't want to execute any other tasks for that - // server (since order might be important). - watchdog_exception('search_api', $e); - $failing_servers[$task->server_id] = TRUE; - } - } - - // Delete all successfully executed tasks. - if ($executed_tasks) { - $this->delete($executed_tasks); - } - // Return TRUE if no tasks failed (i.e., if we didn't mark any server as - // failing) and we retrieved all of them. - return !$failing_servers && $total_count <= $fetch_count; - } + public static function getSubscribedEvents() { + $events = array(); - /** - * Creates a SELECT query for retrieving tasks. - * - * @param \Drupal\search_api\ServerInterface|null $server - * (optional) The server for which to retrieve tasks, or NULL to retrieve - * tasks for all servers. - * - * @return \Drupal\Core\Database\Query\SelectInterface - * A SELECT query for retrieving the tasks. - */ - protected function createQuery(ServerInterface $server = NULL) { - $select = $this->database->select('search_api_task', 't'); - $select->fields('t'); - // Only retrieve tasks we can handle. - $select->condition('t.type', $this->getSupportedTypes(), 'IN'); - if ($server) { - $select->condition('t.server_id', $server->id()); + foreach (static::getSupportedTypes() as $type) { + $events['search_api.task.' . $type][] = array('processEvent'); } - else { - // By ordering by the server, we can later just load them when we reach - // them while looping through the tasks. It is very unlikely there will be - // tasks for more than one or two servers, so a *_load_multiple() probably - // wouldn't bring any significant advantages, but complicate the code. - $select->orderBy('t.server_id'); - } - - // Sometimes the order of tasks might be important, so make sure to order by - // the task ID (which should be in order of insertion). - $select->orderBy('t.id'); - - // Limit number of tasks executed in one call of this method, to avoid - // breaking the site completely if there are too many tasks. - $select->range(0, 100); - // Add a tag and metadata in case other modules want to modify this query. - // The "search_api_server_tasks_limit" metadata is provided for modules that - // change the limit to update it to the new one, so we know about it. - $select->addTag('search_api_server_tasks') - ->addMetaData('search_api_server', $server) - ->addMetaData('search_api_server_tasks_limit', 100); - - // Make it easier for a subclass to change the query. - $this->preprocessQuery($select, $server); - return $select; + return $events; } /** @@ -175,7 +63,7 @@ protected function createQuery(ServerInterface $server = NULL) { * @return string[] * The task types supported by this task manager. */ - protected function getSupportedTypes() { + protected static function getSupportedTypes() { return array( 'addIndex', 'updateIndex', @@ -186,27 +74,58 @@ protected function getSupportedTypes() { } /** - * Preprocesses the query used to retrieve tasks. - * - * Does nothing by default, but can be overridden by subclasses. + * {@inheritdoc} + */ + public function getCount(ServerInterface $server = NULL) { + return $this->taskManager->getTasksCount($this->getTaskConditions($server)); + } + + /** + * {@inheritdoc} + */ + public function execute(ServerInterface $server = NULL) { + if ($server && !$server->status()) { + return FALSE; + } + + $conditions = $this->getTaskConditions($server); + try { + return $this->taskManager->executeAllTasks($conditions, 100); + } + catch (SearchApiException $e) { + watchdog_exception('search_api', $e); + return FALSE; + } + } + + /** + * Processes a single server task. * - * @param \Drupal\Core\Database\Query\SelectInterface $select - * The constructed SELECT query. - * @param \Drupal\search_api\ServerInterface|null $server - * (optional) The server for which tasks are retrieved, or NULL if tasks - * for all servers are retrieved. + * @param \Drupal\search_api\Task\TaskEvent $event + * The task event. */ - protected function preprocessQuery(SelectInterface $select, ServerInterface $server = NULL) {} + public function processEvent(TaskEvent $event) { + $event->stopPropagation(); + + $task = $event->getTask(); + + try { + if (!$this->executeTask($task)) { + $args['%type'] = $task->getType(); + $message = new FormattableMarkup('Task of unknown type %type passed to server task manager.', $args); + throw new SearchApiException($message); + } + } + catch (SearchApiException $e) { + $event->setException($e); + } + } /** * Executes a single server task. * - * @param object $task - * A task object, as retrieved from the search_api_task table. - * @param \Drupal\search_api\ServerInterface $server - * The server on which to execute the task. - * @param \Drupal\search_api\IndexInterface|null $index - * (optional) The index to which the task pertains, if any. + * @param \Drupal\search_api\Task\TaskInterface $task + * The task to execute. * * @return bool * TRUE if the task was successfully executed, FALSE if the task type was @@ -215,8 +134,12 @@ protected function preprocessQuery(SelectInterface $select, ServerInterface $ser * @throws \Drupal\search_api\SearchApiException * If any error occurred while executing the task. */ - protected function executeTask($task, ServerInterface $server, IndexInterface $index = NULL) { - switch ($task->type) { + protected function executeTask(TaskInterface $task) { + $server = $task->getServer(); + $index = $task->getIndex(); + $data = $task->getData(); + + switch ($task->getType()) { case 'addIndex': if ($index) { $server->getBackend()->addIndex($index); @@ -225,24 +148,23 @@ protected function executeTask($task, ServerInterface $server, IndexInterface $i case 'updateIndex': if ($index) { - if ($task->data) { - $index->original = unserialize($task->data); + if ($data) { + $index->original = $data; } $server->getBackend()->updateIndex($index); } return TRUE; case 'removeIndex': + $index = $index ?: $data; if ($index) { - $server->getBackend()->removeIndex($index ? $index : $task->index_id); - $this->delete(NULL, $server, $index); + $server->getBackend()->removeIndex($index); } return TRUE; case 'deleteItems': if ($index && !$index->isReadOnly()) { - $ids = unserialize($task->data); - $server->getBackend()->deleteItems($index, $ids); + $server->getBackend()->deleteItems($index, $data); } return TRUE; @@ -261,139 +183,38 @@ protected function executeTask($task, ServerInterface $server, IndexInterface $i * {@inheritdoc} */ public function setExecuteBatch(ServerInterface $server = NULL) { - $batch_definition = array( - 'operations' => array( - array(array($this, 'processBatch'), array($server ? $server->id() : NULL)), - ), - 'finished' => array($this, 'finishBatch'), - ); - // Schedule the batch. - batch_set($batch_definition); - } - - /** - * Executes server tasks as part of a batch operation. - * - * @param string|null $server_id - * The ID of the server for which tasks should be executed, or NULL to - * execute tasks on all servers. - * @param array $context - * The current batch context, as defined in the @link batch Batch operations - * @endlink documentation. - */ - public function processBatch($server_id, array &$context) { - /** @var \Drupal\search_api\IndexInterface $server */ - $server = $server_id ? $this->loadServer($server_id) : $server_id; - - if (!isset($context['results']['total'])) { - $context['results']['executed'] = 0; - $context['results']['total'] = $this->getCount($server); - if (!$context['results']['total']) { - $context['finished'] = 1; - return; - } - } - - $finished = $this->execute($server); - $still_pending = $this->getCount($server); - $executed = $context['results']['total'] - $still_pending; - if ($finished || $executed == $context['results']['executed']) { - $context['finished'] = 1; - } - else { - $context['finished'] = $still_pending / $context['results']['total']; - $context['message'] = $this->formatPlural($executed, 'Executed @count task.', 'Executed @count tasks.'); - } - $context['results']['executed'] = $executed; - } - - /** - * Finishes an "execute server tasks" batch. - * - * @param bool $success - * Indicates whether the batch process was successful. - * @param array $results - * Results information passed from the processing callback. - */ - public function finishBatch($success, $results) { - // Check if the batch job was successful. - if ($success) { - if ($results['executed']) { - // Display the number of tasks executed. - $message = $this->formatPlural($results['executed'], 'Successfully executed @count task.', 'Successfully executed @count tasks.'); - drupal_set_message($message); - } - if (!$results['total']) { - drupal_set_message($this->t('There were no tasks to execute.'), 'warning'); - } - elseif ($results['executed'] < $results['total']) { - drupal_set_message($this->t('An error occurred while trying to execute the tasks. Check the logs for details.'), 'error'); - } - } - else { - // Notify the user about the batch job failure. - drupal_set_message($this->t('An error occurred while trying to execute the tasks. Check the logs for details.'), 'error'); - } - } - - /** - * {@inheritdoc} - */ - public function add(ServerInterface $server, $type, IndexInterface $index = NULL, $data = NULL) { - $this->database->insert('search_api_task') - ->fields(array( - 'server_id' => $server->id(), - 'type' => $type, - 'index_id' => $index ? (is_object($index) ? $index->id() : $index) : NULL, - 'data' => isset($data) ? serialize($data) : NULL, - )) - ->execute(); + $this->taskManager->setTasksBatch($this->getTaskConditions($server)); } /** * {@inheritdoc} */ - public function delete(array $ids = NULL, ServerInterface $server = NULL, $index = NULL, array $types = NULL) { - $delete = $this->database->delete('search_api_task'); - if ($ids) { - $delete->condition('id', $ids, 'IN'); - } - if ($server) { - $delete->condition('server_id', $server->id()); - } - if ($index) { - $delete->condition('index_id', $index instanceof IndexInterface ? $index->id() : $index); + public function delete(ServerInterface $server = NULL, $index = NULL, array $types = NULL) { + $conditions = $this->getTaskConditions($server); + if ($index !== NULL) { + $conditions['index_id'] = $index instanceof IndexInterface ? $index->id() : $index; } - if ($types) { - $delete->condition('type', $types, 'IN'); + if ($types !== NULL) { + $conditions['type'] = $types; } - $delete->execute(); + $this->taskManager->deleteTasks($conditions); } /** - * Loads a search server. + * Gets a set of conditions for finding the tracking tasks of the given index. * - * @param string $server_id - * The server's ID. - * - * @return \Drupal\search_api\ServerInterface|null - * The loaded server, or NULL if it could not be loaded. - */ - protected function loadServer($server_id) { - return $this->entityTypeManager->getStorage('search_api_server')->load($server_id); - } - - /** - * Loads a search index. - * - * @param string $index_id - * The index's ID. + * @param \Drupal\search_api\ServerInterface $server + * The server for which to retrieve tasks. * - * @return \Drupal\search_api\IndexInterface|null - * The loaded index, or NULL if it could not be loaded. + * @return array + * An array of conditions to pass to the Search API task manager. */ - protected function loadIndex($index_id) { - return $this->entityTypeManager->getStorage('search_api_index')->load($index_id); + protected function getTaskConditions(ServerInterface $server = NULL) { + $conditions['type'] = static::getSupportedTypes(); + if ($server) { + $conditions['server_id'] = $server->id(); + } + return $conditions; } } diff --git a/src/Task/ServerTaskManagerInterface.php b/src/Task/ServerTaskManagerInterface.php index c6adc39..5f9031a 100644 --- a/src/Task/ServerTaskManagerInterface.php +++ b/src/Task/ServerTaskManagerInterface.php @@ -2,7 +2,6 @@ namespace Drupal\search_api\Task; -use Drupal\search_api\IndexInterface; use Drupal\search_api\ServerInterface; /** @@ -46,26 +45,8 @@ public function execute(ServerInterface $server = NULL); public function setExecuteBatch(ServerInterface $server = NULL); /** - * Adds an entry into a server's list of pending tasks. - * - * @param \Drupal\search_api\ServerInterface $server - * The server for which a task should be remembered. - * @param string $type - * The type of task to perform. - * @param \Drupal\search_api\IndexInterface|string|null $index - * (optional) If applicable, the index to which the task pertains (or its - * ID). - * @param mixed $data - * (optional) If applicable, some further data necessary for the task. - */ - public function add(ServerInterface $server, $type, IndexInterface $index = NULL, $data = NULL); - - /** * Removes pending server tasks from the list. * - * @param array|null $ids - * (optional) The IDs of the pending server tasks to delete. Set to NULL - * to not filter by IDs. * @param \Drupal\search_api\ServerInterface|null $server * (optional) A server for which the tasks should be deleted. Set to NULL to * delete tasks from all servers. @@ -76,6 +57,6 @@ public function add(ServerInterface $server, $type, IndexInterface $index = NULL * (optional) The types of tasks that should be deleted, or NULL to delete * tasks regardless of type. */ - public function delete(array $ids = NULL, ServerInterface $server = NULL, $index = NULL, array $types = NULL); + public function delete(ServerInterface $server = NULL, $index = NULL, array $types = NULL); } diff --git a/src/Task/TaskEvent.php b/src/Task/TaskEvent.php new file mode 100644 index 0000000..5d55091 --- /dev/null +++ b/src/Task/TaskEvent.php @@ -0,0 +1,77 @@ +task = $task; + } + + /** + * Retrieves the executed task. + * + * @return \Drupal\search_api\Task\TaskInterface + * The task being executed. + */ + public function getTask() { + return $this->task; + } + + /** + * Retrieves any exception that stopped the execution of the task. + * + * @return \Drupal\search_api\SearchApiException|null + * The exception, if any. + */ + public function getException() { + return $this->exception; + } + + /** + * Sets the exception that stopped the execution of the task. + * + * @param \Drupal\search_api\SearchApiException|null $exception + * The exception that occurred. + * + * @return $this + */ + public function setException(SearchApiException $exception = NULL) { + $this->exception = $exception; + return $this; + } + +} diff --git a/src/Task/TaskInterface.php b/src/Task/TaskInterface.php new file mode 100644 index 0000000..5512da0 --- /dev/null +++ b/src/Task/TaskInterface.php @@ -0,0 +1,85 @@ +entityTypeManager = $entity_type_manager; + $this->eventDispatcher = $event_dispatcher; + $this->setStringTranslation($translation); + } + + /** + * Returns the entity storage for search tasks. + * + * @return \Drupal\Core\Entity\EntityStorageInterface + * The storage handler. + */ + protected function getTaskStorage() { + return $this->entityTypeManager->getStorage('search_api_task'); + } + + /** + * Creates an entity query matching the given search tasks. + * + * @param array $conditions + * (optional) An array of conditions to be matched for the tasks, with + * property names keyed to the value (or values, for multiple possibilities) + * that the property should have. + * + * @return \Drupal\Core\Entity\Query\QueryInterface + * An entity query for search tasks. + */ + protected function getTasksQuery(array $conditions = array()) { + $query = $this->getTaskStorage()->getQuery(); + foreach ($conditions as $property => $values) { + $query->condition($property, (array) $values, 'IN'); + } + $query->sort('id'); + return $query; + } + + /** + * {@inheritdoc} + */ + public function getTasksCount(array $conditions = array()) { + return $this->getTasksQuery($conditions)->count()->execute(); + } + + /** + * {@inheritdoc} + */ + public function addTask($type, ServerInterface $server = NULL, IndexInterface $index = NULL, $data = NULL) { + $task = $this->getTaskStorage()->create(array( + 'type' => $type, + 'server_id' => $server ? $server->id() : NULL, + 'index_id' => $index ? $index->id() : NULL, + 'data' => isset($data) ? serialize($data) : NULL, + )); + $task->save(); + return $task; + } + + /** + * {@inheritdoc} + */ + public function loadTasks(array $conditions = array()) { + $task_ids = $this->getTasksQuery($conditions)->execute(); + if ($task_ids) { + return $this->getTaskStorage()->loadMultiple($task_ids); + } + return array(); + } + + /** + * {@inheritdoc} + */ + public function deleteTask($task_id) { + $task = $this->getTaskStorage()->load($task_id); + if ($task) { + $task->delete(); + } + } + + /** + * {@inheritdoc} + */ + public function deleteTasks(array $conditions = array()) { + $storage = $this->getTaskStorage(); + while (TRUE) { + $task_ids = $this->getTasksQuery($conditions) + ->range(0, 100) + ->execute(); + if (!$task_ids) { + break; + } + $tasks = $storage->loadMultiple($task_ids); + $storage->delete($tasks); + if (count($task_ids) < 100) { + break; + } + } + } + + /** + * {@inheritdoc} + */ + public function executeSpecificTask(TaskInterface $task) { + $event = new TaskEvent($task); + $this->eventDispatcher->dispatch('search_api.task.' . $task->getType(), $event); + if (!$event->isPropagationStopped()) { + $args['%id'] = $task->id(); + $args['%type'] = $task->getType(); + throw new SearchApiException(new FormattableMarkup('Could not execute task %id of type %type. Type seems to be unknown.', $args)); + } + if ($exception = $event->getException()) { + throw $exception; + } + $task->delete(); + } + + /** + * {@inheritdoc} + */ + public function executeSingleTask(array $conditions = array()) { + $task_id = $this->getTasksQuery($conditions)->range(0, 1)->execute(); + if ($task_id) { + $task_id = reset($task_id); + /** @var \Drupal\search_api\Task\TaskInterface $task */ + $task = $this->getTaskStorage()->load($task_id); + $this->executeSpecificTask($task); + return TRUE; + } + return FALSE; + } + + /** + * {@inheritdoc} + */ + public function executeAllTasks(array $conditions = array(), $limit = NULL) { + // We have to use this roundabout way because tasks, during their execution, + // might create additional tasks. (E.g., see + // \Drupal\search_api\Task\IndexTaskManager::trackItems().) + $executed = 0; + while (TRUE) { + $query = $this->getTasksQuery($conditions); + if (isset($limit)) { + $query->range(0, $limit - $executed); + } + $task_ids = $query->execute(); + + if (!$task_ids) { + break; + } + + foreach ($task_ids as $task_id) { + /** @var \Drupal\search_api\Task\TaskInterface $task */ + $task = $this->getTaskStorage()->load($task_id); + if ($task) { + $this->executeSpecificTask($task); + } + else { + --$executed; + } + } + + $executed += count($task_ids); + if (isset($limit) && $executed >= $limit) { + break; + } + } + + return !$this->getTasksCount($conditions); + } + + /** + * {@inheritdoc} + */ + public function setTasksBatch(array $conditions = array()) { + $task_ids = $this->getTasksQuery($conditions)->range(0, 100)->execute(); + + if (!$task_ids) { + return; + } + + $batch_definition = array( + 'operations' => array( + array(array($this, 'processBatch'), array($task_ids, $conditions)), + ), + 'finished' => array($this, 'finishBatch'), + ); + // Schedule the batch. + batch_set($batch_definition); + } + + /** + * Processes a single pending task as part of a batch operation. + * + * @param int[] $task_ids + * An array of task IDs to execute. Might not contain all task IDs. + * @param array $conditions + * An array of conditions defining the tasks to be executed. Should be used + * to retrieve more task IDs if necessary. + * @param array $context + * The current batch context, as defined in the @link batch Batch operations + * @endlink documentation. + * + * @throws \Drupal\search_api\SearchApiException + * Thrown if any error occurred while processing the task. + */ + public function processBatch(array $task_ids, array $conditions, array &$context) { + // Initialize context information. + if (!isset($context['task_ids'])) { + $context['task_ids'] = $task_ids; + $context['conditions'] = $conditions; + $context['results']['total'] = $this->getTasksCount($conditions); + } + + $task_id = array_shift($context['task_ids']); + /** @var \Drupal\search_api\Task\TaskInterface $task */ + $task = $this->getTaskStorage()->load($task_id); + + $this->executeSpecificTask($task); + + if (!$context['task_ids']) { + $context['task_ids'] = $this->getTasksQuery($conditions) + ->range(0, 100) + ->execute(); + if (!$context['task_ids']) { + $context['finished'] = 1; + return; + } + } + + $pending = $this->getTasksCount($conditions); + $context['finished'] = 1 - $pending / $context['results']['total']; + $executed = $context['results']['total'] - $pending; + $context['message'] = $this->formatPlural( + $executed, + 'Successfully executed @count pending task.', + 'Successfully executed @count pending tasks.' + ); + } + + /** + * Finishes an "execute tasks" batch. + * + * @param bool $success + * Indicates whether the batch process was successful. + * @param array $results + * Results information passed from the processing callback. + */ + public function finishBatch($success, array $results) { + // Check if the batch job was successful. + if ($success) { + $message = $this->formatPlural( + $results['total'], + 'Successfully executed @count pending task.', + 'Successfully executed @count pending tasks.' + ); + drupal_set_message($message); + } + else { + // Notify the user about the batch job failure. + drupal_set_message($this->t('An error occurred while trying to execute tasks. Check the logs for details.'), 'error'); + } + } + +} diff --git a/src/Task/TaskManagerInterface.php b/src/Task/TaskManagerInterface.php new file mode 100644 index 0000000..f84f027 --- /dev/null +++ b/src/Task/TaskManagerInterface.php @@ -0,0 +1,133 @@ +get('search_api.server_task_manager') - ->add($server, 'deleteItems', $index, array('')); + ->get('search_api.task_manager') + ->addTask('deleteItems', $server, $index, array('')); // Due to an (apparent) Core bug we need to clear the cache, otherwise the // "local actions" block gets displayed from cache (without the link). See // #2722237. diff --git a/src/Utility.php b/src/Utility.php index 70b52df..f1488be 100644 --- a/src/Utility.php +++ b/src/Utility.php @@ -444,6 +444,9 @@ public static function isFieldIdReserved($field_id) { * The current batch context. * @param \Drupal\Core\Config\ConfigImporter $config_importer * The config importer. + * + * @throws \Drupal\search_api\SearchApiException + * Thrown if any error occurred while tracking items. */ public static function processIndexTasks(array &$context, ConfigImporter $config_importer) { $index_task_manager = \Drupal::getContainer()->get('search_api.index_task_manager'); @@ -471,11 +474,15 @@ public static function processIndexTasks(array &$context, ConfigImporter $config } } - $index_id = array_shift($context['sandbox']['indexes']); - $index = Index::load($index_id); - $added = $index_task_manager->addItemsOnce($index); - if ($added !== NULL) { - array_unshift($context['sandbox']['indexes'], $index_id); + try { + $index_id = array_shift($context['sandbox']['indexes']); + $index = Index::load($index_id); + if (!($index_task_manager->addItemsOnce($index))) { + array_unshift($context['sandbox']['indexes'], $index_id); + } + } + catch (SearchApiException $e) { + watchdog_exception('search_api', $e); } if (empty($context['sandbox']['indexes'])) { diff --git a/tests/src/Kernel/CliTest.php b/tests/src/Kernel/CliTest.php index 50aa4c7..f76c466 100644 --- a/tests/src/Kernel/CliTest.php +++ b/tests/src/Kernel/CliTest.php @@ -70,14 +70,14 @@ public function testItemTracking() { EntityTest::create(array( 'name' => 'foo bar baz föö smile' . json_decode('"\u1F601"'), 'body' => 'test test case Case casE', - 'type' => 'item', + 'type' => 'entity_test', 'keywords' => array('Orange', 'orange', 'örange', 'Orange'), 'category' => 'item_category', ))->save(); EntityTest::create(array( 'name' => 'foo bar baz föö smile', 'body' => 'test test case Case casE', - 'type' => 'item', + 'type' => 'entity_test', 'keywords' => array('strawberry', 'llama'), 'category' => 'item_category', ))->save(); diff --git a/tests/src/Kernel/DependencyRemovalTest.php b/tests/src/Kernel/DependencyRemovalTest.php index d879472..b2b8f7b 100644 --- a/tests/src/Kernel/DependencyRemovalTest.php +++ b/tests/src/Kernel/DependencyRemovalTest.php @@ -48,11 +48,7 @@ class DependencyRemovalTest extends KernelTestBase { public function setUp() { parent::setUp(); - $this->installSchema('system', 'key_value_expire'); - - // The server tasks manager is needed when removing a server. - $mock = $this->getMock('Drupal\search_api\Task\ServerTaskManagerInterface'); - $this->container->set('search_api.server_task_manager', $mock); + $this->installSchema('search_api', 'search_api_task'); // Create the index object, but don't save it yet since we want to change // its settings anyways in every test. diff --git a/tests/src/Kernel/IndexStorageTest.php b/tests/src/Kernel/IndexStorageTest.php index 32c6e0e..4f590ab 100644 --- a/tests/src/Kernel/IndexStorageTest.php +++ b/tests/src/Kernel/IndexStorageTest.php @@ -34,6 +34,7 @@ protected function setUp() { parent::setUp(); $this->installSchema('system', 'key_value_expire'); + $this->installSchema('search_api', 'search_api_task'); $this->storage = $this->container ->get('entity_type.manager') diff --git a/tests/src/Kernel/Processor/ContentAccessTest.php b/tests/src/Kernel/Processor/ContentAccessTest.php index bfbd888..e8e3521 100644 --- a/tests/src/Kernel/Processor/ContentAccessTest.php +++ b/tests/src/Kernel/Processor/ContentAccessTest.php @@ -44,10 +44,6 @@ class ContentAccessTest extends ProcessorTestBase { public function setUp($processor = NULL) { parent::setUp('content_access'); - // The parent method already installs most needed node and comment schemas, - // but here we also need the comment statistics. - $this->installSchema('comment', array('comment_entity_statistics')); - // Create a node type for testing. $type = NodeType::create(array('type' => 'page', 'name' => 'page')); $type->save(); diff --git a/tests/src/Kernel/Processor/ProcessorTestBase.php b/tests/src/Kernel/Processor/ProcessorTestBase.php index 45188b7..4ae3aff 100644 --- a/tests/src/Kernel/Processor/ProcessorTestBase.php +++ b/tests/src/Kernel/Processor/ProcessorTestBase.php @@ -69,6 +69,7 @@ public function setUp($processor = NULL) { $this->installEntitySchema('user'); $this->installEntitySchema('node'); $this->installEntitySchema('comment'); + $this->installSchema('comment', array('comment_entity_statistics')); $this->installConfig(array('field')); Action::create([ diff --git a/tests/src/Kernel/ServerTaskTest.php b/tests/src/Kernel/ServerTaskTest.php index 2c589a5..5426e10 100644 --- a/tests/src/Kernel/ServerTaskTest.php +++ b/tests/src/Kernel/ServerTaskTest.php @@ -56,6 +56,13 @@ class ServerTaskTest extends KernelTestBase { protected $state; /** + * The task manager to use for the tests. + * + * @var \Drupal\search_api\Task\TaskManagerInterface + */ + protected $taskManager; + + /** * The server task manager to use for the tests. * * @var \Drupal\search_api\Task\ServerTaskManagerInterface @@ -104,6 +111,7 @@ public function setUp() { )); $this->index->save(); + $this->taskManager = $this->container->get('search_api.task_manager'); $this->serverTaskManager = $this->container->get('search_api.server_task_manager'); // Reset the list of called backend methods. @@ -339,7 +347,7 @@ public function testDeleteAllIndexItems() { public function testTaskCountLimit() { // Create 100 tasks. for ($i = 0; $i < 101; ++$i) { - $this->serverTaskManager->add($this->server, 'deleteItems', $this->index, array('')); + $this->taskManager->addTask('deleteItems', $this->server, $this->index, array('')); } // Verify that a new operation cannot be executed. @@ -364,7 +372,7 @@ public function testAutomaticTaskRemoval() { 'backend' => 'search_api_test_backend', )); $server2->save(); - $this->serverTaskManager->add($server2, 'removeIndex', $this->index); + $this->taskManager->addTask('removeIndex', $server2, $this->index); $index_values = $this->index->toArray(); unset($index_values['uuid']); @@ -424,12 +432,12 @@ public function testAutomaticTaskRemoval() { * A second index, for which one additional "addIndex" task is created. */ protected function addTasks(IndexInterface $second_index) { - $this->serverTaskManager->add($this->server, 'addIndex', $second_index); - $this->serverTaskManager->add($this->server, 'removeIndex', $this->index); - $this->serverTaskManager->add($this->server, 'addIndex', $this->index); - $this->serverTaskManager->add($this->server, 'updateIndex', $this->index); - $this->serverTaskManager->add($this->server, 'deleteItems', $this->index, array()); - $this->serverTaskManager->add($this->server, 'deleteAllIndexItems', $this->index); + $this->taskManager->addTask('addIndex', $this->server, $second_index); + $this->taskManager->addTask('removeIndex', $this->server, $this->index); + $this->taskManager->addTask('addIndex', $this->server, $this->index); + $this->taskManager->addTask('updateIndex', $this->server, $this->index); + $this->taskManager->addTask('deleteItems', $this->server, $this->index, array()); + $this->taskManager->addTask('deleteAllIndexItems', $this->server, $this->index); } /**