git.aero2k.de Git - tmp/jakarta-migration.git/blob

   1 /*
   2  * This file contains code from Florent Guillame's nuxeo-reindex-fulltext module.
   3  *
   4  */
   5
   6 package org.collectionspace.services.batch.nuxeo;
   7
   8 import java.io.File;
   9 import java.io.Serializable;
  10 import java.lang.reflect.Field;
  11 import java.net.URISyntaxException;
  12
  13 import java.util.ArrayList;
  14 import java.util.Arrays;
  15 import java.util.Collections;
  16 import java.util.HashMap;
  17 import java.util.HashSet;
  18 import java.util.LinkedHashSet;
  19 import java.util.List;
  20 import java.util.Map;
  21 import java.util.Set;
  22
  23 import org.apache.commons.lang.StringUtils;
  24 import org.collectionspace.services.batch.BatchCommon;
  25 import org.collectionspace.services.common.CollectionSpaceResource;
  26 import org.collectionspace.services.common.NuxeoBasedResource;
  27 import org.collectionspace.services.common.StoredValuesUriTemplate;
  28 import org.collectionspace.services.common.UriTemplateFactory;
  29 import org.collectionspace.services.common.UriTemplateRegistryKey;
  30 import org.collectionspace.services.common.invocable.InvocationContext.ListCSIDs;
  31 import org.collectionspace.services.common.invocable.InvocationContext.Params.Param;
  32 import org.collectionspace.services.common.invocable.InvocationResults;
  33 import org.collectionspace.services.common.vocabulary.AuthorityResource;
  34
  35 import org.dom4j.DocumentException;
  36 import org.nuxeo.ecm.core.api.AbstractSession;
  37 import org.nuxeo.ecm.core.api.CoreSession;
  38 import org.nuxeo.ecm.core.api.IterableQueryResult;
  39 import org.nuxeo.ecm.core.api.NuxeoException;
  40 import org.nuxeo.ecm.core.event.EventService;
  41 import org.nuxeo.ecm.core.query.QueryFilter;
  42 import org.nuxeo.ecm.core.query.sql.NXQL;
  43 import org.nuxeo.ecm.core.storage.FulltextConfiguration;
  44 import org.nuxeo.ecm.core.storage.sql.Model;
  45 import org.nuxeo.ecm.core.storage.sql.Node;
  46 import org.nuxeo.ecm.core.storage.sql.Session;
  47 import org.nuxeo.ecm.core.storage.sql.SimpleProperty;
  48 import org.nuxeo.ecm.core.storage.sql.coremodel.SQLFulltextExtractorWork;
  49 import org.nuxeo.ecm.core.storage.sql.coremodel.SQLSession;
  50 import org.nuxeo.ecm.core.work.api.Work;
  51 import org.nuxeo.ecm.core.work.api.WorkManager;
  52 import org.nuxeo.ecm.core.work.api.WorkManager.Scheduling;
  53 import org.nuxeo.runtime.api.Framework;
  54 import org.nuxeo.runtime.transaction.TransactionHelper;
  55 import org.slf4j.Logger;
  56 import org.slf4j.LoggerFactory;
  57
  58 public class ReindexFullTextBatchJob extends AbstractBatchJob {
  59         final Logger log = LoggerFactory.getLogger(ReindexFullTextBatchJob.class);
  60
  61         public static final String DC_TITLE = "dc:title";
  62         public static final int DEFAULT_BATCH_SIZE = 1000;
  63         public static final int DEFAULT_START_BATCH = 0;
  64         public static final int DEFAULT_END_BATCH = 0;
  65         public static final int DEFAULT_BATCH_PAUSE = 0;
  66         public static final String BATCH_STOP_FILE = "stopBatch";
  67         public static final String DOCTYPE_STOP_FILE = "stopDocType";
  68
  69         private int batchSize = DEFAULT_BATCH_SIZE;
  70         private int batchPause = DEFAULT_BATCH_PAUSE;
  71         private int startBatch = DEFAULT_START_BATCH;
  72         private int endBatch = DEFAULT_END_BATCH;
  73         private int numAffected = 0;
  74
  75         private String stopFileDirectory;
  76
  77         private CoreSession coreSession;
  78         private Session session = null;
  79     protected FulltextConfiguration fulltextConfiguration;
  80
  81         private Map<String, NuxeoBasedResource> resourcesByDocType;
  82
  83         public ReindexFullTextBatchJob() {
  84                 setSupportedInvocationModes(Arrays.asList(INVOCATION_MODE_NO_CONTEXT, INVOCATION_MODE_SINGLE, INVOCATION_MODE_LIST));
  85
  86                 stopFileDirectory = System.getProperty("java.io.tmpdir") + File.separator + ReindexFullTextBatchJob.class.getName();
  87
  88                 log.debug("stop file directory is " + stopFileDirectory);
  89         }
  90
  91         //
  92         // Since the ReindexFullTextBatchJob class deals with transactions differently than other batch jobs, we need to
  93         // override this method to ensure there is an active transaction.
  94         //
  95         @Override
  96         protected List<String> getVocabularyCsids(AuthorityResource<?, ?> resource) throws URISyntaxException {
  97                 boolean tx = false;
  98                 if (TransactionHelper.isTransactionActive() == false) {
  99                         tx = TransactionHelper.startTransaction();
 100                 }
 101
 102                 try {
 103                         return super.getVocabularyCsids(resource);
 104                 } finally {
 105                         if (tx) {
 106                                 TransactionHelper.commitOrRollbackTransaction();
 107                         }
 108                 }
 109         }
 110
 111         //
 112         // Since the ReindexFullTextBatchJob class deals with transactions differently than other batch jobs, we need to
 113         // override this method to ensure there is an active transaction.
 114         //
 115         @Override
 116         protected List<String> findAll(NuxeoBasedResource resource, int pageSize, int pageNum, String sortBy)
 117                         throws URISyntaxException, DocumentException {
 118                 boolean tx = false;
 119                 if (TransactionHelper.isTransactionActive() == false) {
 120                         tx = TransactionHelper.startTransaction();
 121                 }
 122
 123                 try {
 124                         return super.findAll(resource, pageSize, pageNum, sortBy);
 125                 } finally {
 126                         if (tx) {
 127                                 TransactionHelper.commitOrRollbackTransaction();
 128                         }
 129                 }
 130         }
 131
 132         //
 133         // Since the ReindexFullTextBatchJob class deals with transactions differently than other batch jobs, we need to
 134         // override this method to ensure there is an active transaction.
 135         //
 136         @Override
 137         protected List<String> findAllAuthorityItems(AuthorityResource<?, ?> resource, String vocabularyCsid, int pageSize, int pageNum, String sortBy)
 138                         throws URISyntaxException, Exception {
 139                 boolean tx = false;
 140                 if (TransactionHelper.isTransactionActive() == false) {
 141                         tx = TransactionHelper.startTransaction();
 142                 }
 143
 144                 try {
 145                         return super.findAllAuthorityItems(resource, vocabularyCsid, pageSize, pageNum, sortBy);
 146                 } finally {
 147                         if (tx) {
 148                                 TransactionHelper.commitOrRollbackTransaction();
 149                         }
 150                 }
 151         }
 152
 153
 154         @Override
 155         public void run() {
 156                 run(null);
 157         }
 158
 159         @Override
 160         public void run(BatchCommon batchCommon) {
 161                 setCompletionStatus(STATUS_MIN_PROGRESS);
 162
 163                 numAffected = 0;
 164
 165                 boolean isTransactionActive = TransactionHelper.isTransactionActive();
 166
 167                 // Commit and close the transaction that was started by the standard request lifecycle.
 168
 169                 if (isTransactionActive) {
 170                         TransactionHelper.commitOrRollbackTransaction();
 171                 }
 172
 173                 try {
 174                         coreSession = getRepoSession().getCoreSession();
 175
 176                         if (requestIsForInvocationModeSingle()) {
 177                                 String csid = getInvocationContext().getSingleCSID();
 178
 179                                 if (csid == null) {
 180                                         throw new Exception("No singleCSID was supplied in invocation context.");
 181                                 }
 182
 183                                 String docType = getInvocationContext().getDocType();
 184
 185                                 if (StringUtils.isEmpty(docType)) {
 186                                         throw new Exception("No docType was supplied in invocation context.");
 187                                 }
 188
 189                                 log.debug("Reindexing " + docType + " record with csid: " + csid);
 190
 191                                 reindexDocument(docType, csid);
 192                         }
 193                         else if (requestIsForInvocationModeList()) {
 194                                 ListCSIDs list = getInvocationContext().getListCSIDs();
 195                                 List<String> csids = list.getCsid();
 196
 197                                 if (csids == null || csids.size() == 0) {
 198                                         throw new Exception("no listCSIDs were supplied");
 199                                 }
 200
 201                                 String docType = getInvocationContext().getDocType();
 202
 203                                 if (StringUtils.isEmpty(docType)) {
 204                                         throw new Exception("No docType was supplied in invocation context.");
 205                                 }
 206
 207                                 log.debug("Reindexing " + csids.size() + " " + docType + " records with csids: " + csids.get(0) + ", ...");
 208
 209                                 if (log.isTraceEnabled()) {
 210                                         log.trace(StringUtils.join(csids, ", "));
 211                                 }
 212
 213                                 reindexDocuments(docType, csids);
 214                         }
 215                         else if (requestIsForInvocationModeNoContext()) {
 216                                 Set<String> docTypes = new LinkedHashSet<String>();
 217                                 String docType;
 218
 219                                 docType = getInvocationContext().getDocType();
 220
 221                                 if (StringUtils.isNotEmpty(docType)) {
 222                                         docTypes.add(docType);
 223                                 }
 224
 225                                 // Read batch size, start and end batches, pause, and additional doctypes from params.
 226
 227                                 for (Param param : this.getParams()) {
 228                                         if (param.getKey().equals("batchSize")) {
 229                                                 batchSize = Integer.parseInt(param.getValue());
 230                                         }
 231                                         else if (param.getKey().equals("startBatch")) {
 232                                                 startBatch = Integer.parseInt(param.getValue());
 233                                         }
 234                                         else if (param.getKey().equals("endBatch")) {
 235                                                 endBatch = Integer.parseInt(param.getValue());
 236                                         }
 237                                         else if (param.getKey().equals("batchPause")) {
 238                                                 batchPause = Integer.parseInt(param.getValue());
 239                                         }
 240                                         else if (param.getKey().equals("docType")) {
 241                                                 docType = param.getValue();
 242
 243                                                 if (StringUtils.isNotEmpty(docType)) {
 244                                                         docTypes.add(docType);
 245                                                 }
 246                                         }
 247                                 }
 248
 249                                 //
 250                                 // If docTypes is empty, we should use the <forDocTypes> list from the resource/payload
 251                                 //
 252                                 if (docTypes.isEmpty() == true && batchCommon != null) {
 253                                         List<String> payloadDocTypes = batchCommon.getForDocTypes().getForDocType();
 254                                         if (payloadDocTypes != null && !payloadDocTypes.isEmpty()) {
 255                                                 docTypes = convertListToSet(payloadDocTypes);
 256                                         }
 257                                 }
 258
 259                                 initResourceMap();
 260                                 reindexDocuments(docTypes);
 261                         }
 262
 263                         log.debug("reindexing complete");
 264
 265                         InvocationResults results = new InvocationResults();
 266                         results.setNumAffected(numAffected);
 267                         results.setUserNote("reindexed " + numAffected + " records");
 268
 269                         setResults(results);
 270                         setCompletionStatus(STATUS_COMPLETE);
 271                 }
 272                 catch(StoppedException e) {
 273                         log.debug("reindexing terminated by stop file");
 274
 275                         InvocationResults results = new InvocationResults();
 276                         results.setNumAffected(numAffected);
 277                         results.setUserNote("reindexing terminated by stop file");
 278
 279                         setResults(results);
 280                         setCompletionStatus(STATUS_COMPLETE);
 281                 }
 282                 catch(Exception e) {
 283                         setErrorResult(e.getMessage());
 284                 }
 285                 finally {
 286                         // Start a new transaction so the standard request lifecycle can complete.
 287
 288                         if (isTransactionActive) {
 289                                 TransactionHelper.startTransaction();
 290                         }
 291                 }
 292         }
 293
 294         private void initResourceMap() {
 295                 resourcesByDocType = new HashMap<String, NuxeoBasedResource>();
 296
 297                 for (CollectionSpaceResource<?, ?> resource : getResourceMap().values()) {
 298                         Map<UriTemplateRegistryKey, StoredValuesUriTemplate> entries = resource.getUriRegistryEntries();
 299
 300                         for (UriTemplateRegistryKey key : entries.keySet()) {
 301                                 String docType = key.getDocType();
 302                                 String tenantId = key.getTenantId();
 303
 304                                 if (getTenantId().equals(tenantId)) {
 305                                         if (resourcesByDocType.containsKey(docType)) {
 306                                                 log.warn("multiple resources found for docType " + docType);
 307
 308                                                 NuxeoBasedResource currentResource = resourcesByDocType.get(docType);
 309                                                 NuxeoBasedResource candidateResource = (NuxeoBasedResource) resource;
 310
 311                                                 // Favor the resource that isn't an AuthorityResource. This
 312                                                 // is really just to deal with Contacts, which are handled
 313                                                 // by ContactResource, PersonAuthorityResource, and
 314                                                 // OrgAuthorityResource. We want to use ContactResource.
 315
 316                                                 if (!(candidateResource instanceof AuthorityResource) && (currentResource instanceof AuthorityResource)) {
 317                                                         resourcesByDocType.put(docType, candidateResource);
 318                                                 }
 319
 320                                                 log.warn("using " + resourcesByDocType.get(docType));
 321                                         }
 322                                         else {
 323                                                 resourcesByDocType.put(docType, (NuxeoBasedResource) resource);
 324                                         }
 325                                 }
 326                         }
 327                 }
 328         }
 329
 330         private void reindexDocuments(Set<String> docTypes) throws Exception {
 331                 if (docTypes == null) {
 332                         docTypes = new LinkedHashSet<String>();
 333                 }
 334
 335                 // If no types are specified, do them all.
 336
 337                 if (docTypes.size() == 0) {
 338                         docTypes.addAll(getAllDocTypes());
 339                 }
 340
 341                 for (String docType : docTypes) {
 342                         reindexDocuments(docType);
 343                 }
 344         }
 345
 346         private List<String> getAllDocTypes() {
 347                 List<String> docTypes = new ArrayList<String>(resourcesByDocType.keySet());
 348                 Collections.sort(docTypes);
 349
 350                 log.debug("Call to getAllDocTypes() method found: " + StringUtils.join(docTypes, ", "));
 351
 352                 return docTypes;
 353         }
 354
 355         private void reindexDocuments(String docType) throws Exception {
 356                 // Check for a stop file before reindexing the docType.
 357
 358                 if (batchStopFileExists() || docTypeStopFileExists()) {
 359                         throw new StoppedException();
 360                 }
 361
 362                 log.debug("reindexing docType " + docType);
 363
 364                 NuxeoBasedResource resource = resourcesByDocType.get(docType);
 365
 366                 if (resource == null) {
 367                         log.warn("No service resource found for docType " + docType);
 368                 }
 369
 370                 boolean isAuthorityItem = false;
 371
 372                 if (resource instanceof AuthorityResource) {
 373                         UriTemplateRegistryKey key = new UriTemplateRegistryKey(getTenantId(), docType);
 374                         StoredValuesUriTemplate uriTemplate = resource.getUriRegistryEntries().get(key);
 375
 376                         log.debug("uriTemplateType=" + uriTemplate.getUriTemplateType());
 377
 378                         if (uriTemplate.getUriTemplateType() == UriTemplateFactory.ITEM) {
 379                                 isAuthorityItem = true;
 380                         }
 381                 }
 382
 383                 int pageSize = batchSize;
 384
 385                 // The supplied start and end batch numbers start with 1, but the page number starts with 0.
 386                 int startPage = (startBatch > 0) ? startBatch - 1 : 0;
 387                 int endPage = (endBatch > 0) ? endBatch - 1 : Integer.MAX_VALUE;
 388
 389                 if (isAuthorityItem) {
 390                         List<String> vocabularyCsids = getVocabularyCsids((AuthorityResource<?, ?>) resource);
 391
 392                         for (String vocabularyCsid : vocabularyCsids) {
 393                                 int pageNum = startPage;
 394                                 List<String> csids = null;
 395
 396                                 log.debug("Reindexing vocabulary of " + docType + " with csid " + vocabularyCsid);
 397
 398                                 do {
 399                                         // Check for a stop file before reindexing the batch.
 400
 401                                         if (batchStopFileExists()) {
 402                                                 throw new StoppedException();
 403                                         }
 404
 405                                         csids = findAllAuthorityItems((AuthorityResource<?, ?>) resource, vocabularyCsid, pageSize, pageNum, "collectionspace_core:createdAt, ecm:name");
 406
 407                                         if (csids.size() > 0) {
 408                                                 log.debug("reindexing vocabulary of " + docType +" with csid " + vocabularyCsid + ", batch " + (pageNum + 1) + ": " + csids.size() + " records starting with " + csids.get(0));
 409
 410                                                 // Pause for the configured amount of time.
 411
 412                                                 if (batchPause > 0) {
 413                                                         log.trace("pausing " + batchPause + " ms");
 414
 415                                                         Thread.sleep(batchPause);
 416                                                 }
 417
 418                                                 reindexDocuments(docType, csids);
 419                                         }
 420
 421                                         pageNum++;
 422                                 }
 423                                 while(csids.size() == pageSize && pageNum <= endPage);
 424                         }
 425                 } else {
 426                         int pageNum = startPage;
 427                         List<String> csids = null;
 428
 429                         do {
 430                                 // Check for a stop file before reindexing the batch.
 431
 432                                 if (batchStopFileExists()) {
 433                                         throw new StoppedException();
 434                                 }
 435
 436                                 csids = findAll(resource, pageSize, pageNum, "collectionspace_core:createdAt, ecm:name");
 437
 438                                 if (csids.size() > 0) {
 439                                         log.debug("reindexing " + docType +" batch " + (pageNum + 1) + ": " + csids.size() + " records starting with " + csids.get(0));
 440
 441                                         // Pause for the configured amount of time.
 442
 443                                         if (batchPause > 0) {
 444                                                 log.trace("pausing " + batchPause + " ms");
 445
 446                                                 Thread.sleep(batchPause);
 447                                         }
 448
 449                                         reindexDocuments(docType, csids);
 450                                 }
 451
 452                                 pageNum++;
 453                         }
 454                         while(csids.size() == pageSize && pageNum <= endPage);
 455                 }
 456         }
 457
 458         private void reindexDocument(String docType, String csid) throws Exception {
 459                 reindexDocuments(docType, Arrays.asList(csid));
 460         }
 461
 462         private void reindexDocuments(String docType, List<String> csids) throws Exception {
 463                 // Convert the csids to structs of nuxeo id and type, as expected
 464                 // by doBatch.
 465
 466                 if (csids == null || csids.size() == 0) {
 467                         return;
 468                 }
 469
 470                 // Transaction for the batch
 471                 boolean tx = TransactionHelper.startTransaction();
 472
 473                 getLowLevelSession();
 474
 475                 List<Info> infos = new ArrayList<Info>();
 476
 477                 String query = "SELECT ecm:uuid, ecm:primaryType FROM Document " +
 478                                            "WHERE ecm:name IN (" + StringUtils.join(quoteList(csids), ',') + ") " +
 479                                            "AND ecm:primaryType LIKE '" + docType + "%' " +
 480                                            "AND ecm:isCheckedInVersion = 0 AND ecm:isProxy = 0";
 481                 IterableQueryResult result = session.queryAndFetch(query, NXQL.NXQL, QueryFilter.EMPTY);
 482
 483                 try {
 484                         for (Map<String, Serializable> map : result) {
 485                                 String id = (String) map.get(NXQL.ECM_UUID);
 486                                 String type = (String) map.get(NXQL.ECM_PRIMARYTYPE);
 487                                 infos.add(new Info(id, type));
 488                         }
 489                 } finally {
 490                         result.close();
 491                 }
 492
 493                 if (csids.size() != infos.size()) {
 494                         log.warn("didn't find info for all the supplied csids: expected " + csids.size() + ", found " + infos.size());
 495                 }
 496
 497                 if (log.isTraceEnabled()) {
 498                         for (Info info : infos) {
 499                                 log.trace(info.type + " " + info.id);
 500                         }
 501                 }
 502
 503                 numAffected += infos.size();
 504
 505                 // Below code copied from the doBatch function.
 506
 507                 boolean ok;
 508
 509                 List<Serializable> ids = new ArrayList<Serializable>(infos.size());
 510                 Set<String> asyncIds = new HashSet<String>();
 511                 Model model = session.getModel();
 512                 for (Info info : infos) {
 513                         ids.add(info.id);
 514                         if (fulltextConfiguration.isFulltextIndexable(info.type)) {
 515                                 asyncIds.add(model.idToString(info.id));
 516                         }
 517                 }
 518                 ok = false;
 519                 try {
 520                         runSyncBatch(ids, asyncIds);
 521                         ok = true;
 522                 } finally {
 523                         if (tx) {
 524                                 if (!ok) {
 525                                         TransactionHelper.setTransactionRollbackOnly();
 526                                         log.error("Rolling back sync");
 527                                 }
 528                                 TransactionHelper.commitOrRollbackTransaction();
 529                         }
 530                 }
 531
 532                 runAsyncBatch(asyncIds);
 533
 534                 // wait for async completion after transaction commit
 535                 Framework.getLocalService(EventService.class).waitForAsyncCompletion();
 536         }
 537
 538         private List<String> quoteList(List<String> values) {
 539                 List<String> quoted = new ArrayList<String>(values.size());
 540
 541                 for (String value : values) {
 542                         quoted.add("'" + value + "'");
 543                 }
 544
 545                 return quoted;
 546         }
 547
 548         private boolean batchStopFileExists() {
 549                 return (stopFileDirectory != null && new File(stopFileDirectory + File.separator + BATCH_STOP_FILE).isFile());
 550         }
 551
 552         private boolean docTypeStopFileExists() {
 553                 return (stopFileDirectory != null && new File(stopFileDirectory + File.separator + DOCTYPE_STOP_FILE).isFile());
 554         }
 555
 556         private static class StoppedException extends Exception {
 557                 private static final long serialVersionUID = 8813189331855935939L;
 558
 559                 public StoppedException() {
 560
 561                 }
 562         }
 563
 564         /*
 565          * The code below this comment is copied from the nuxeo-reindex-fulltext
 566          * module. The original copyright is below.
 567          */
 568
 569         /*
 570          * (C) Copyright 2012 Nuxeo SA (http://nuxeo.com/) and contributors.
 571          *
 572          * All rights reserved. This program and the accompanying materials
 573          * are made available under the terms of the GNU Lesser General Public License
 574          * (LGPL) version 2.1 which accompanies this distribution, and is available at
 575          * http://www.gnu.org/licenses/lgpl.html
 576          *
 577          * This library is distributed in the hope that it will be useful,
 578          * but WITHOUT ANY WARRANTY; without even the implied warranty of
 579          * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
 580          * Lesser General Public License for more details.
 581          *
 582          * Contributors:
 583          *     Florent Guillaume
 584          */
 585
 586         protected static class Info {
 587                 public final Serializable id;
 588
 589                 public final String type;
 590
 591                 public Info(Serializable id, String type) {
 592                         this.id = id;
 593                         this.type = type;
 594                 }
 595         }
 596
 597         /**
 598          * This has to be called once the transaction has been started.
 599          */
 600         protected void getLowLevelSession() {
 601                 try {
 602                         SQLSession s = (SQLSession) ((AbstractSession) coreSession).getSession();
 603                         Field f2 = SQLSession.class.getDeclaredField("session");
 604                         f2.setAccessible(true);
 605                         session = (Session) f2.get(s);
 606                         fulltextConfiguration = session.getModel().getFulltextConfiguration();
 607                 } catch (ReflectiveOperationException e) {
 608                         throw new NuxeoException(e);
 609                 }
 610         }
 611
 612         protected void doBatch(List<Info> infos) {
 613                 boolean tx;
 614                 boolean ok;
 615
 616                 // transaction for the sync batch
 617                 tx = TransactionHelper.startTransaction();
 618
 619                 getLowLevelSession(); // for fulltextInfo
 620                 List<Serializable> ids = new ArrayList<Serializable>(infos.size());
 621                 Set<String> asyncIds = new HashSet<String>();
 622                 Model model = session.getModel();
 623                 for (Info info : infos) {
 624                         ids.add(info.id);
 625                         if (fulltextConfiguration.isFulltextIndexable(info.type)) {
 626                                 asyncIds.add(model.idToString(info.id));
 627                         }
 628                 }
 629                 ok = false;
 630                 try {
 631                         runSyncBatch(ids, asyncIds);
 632                         ok = true;
 633                 } finally {
 634                         if (tx) {
 635                                 if (!ok) {
 636                                         TransactionHelper.setTransactionRollbackOnly();
 637                                         log.error("Rolling back sync");
 638                                 }
 639                                 TransactionHelper.commitOrRollbackTransaction();
 640                         }
 641                 }
 642
 643                 runAsyncBatch(asyncIds);
 644
 645                 // wait for async completion after transaction commit
 646                 Framework.getLocalService(EventService.class).waitForAsyncCompletion();
 647         }
 648
 649         /*
 650                 * Do this at the low-level session level because we may have to modify
 651                 * things like versions which aren't usually modifiable, and it's also good
 652                 * to bypass all listeners.
 653                 */
 654         protected void runSyncBatch(List<Serializable> ids, Set<String> asyncIds) {
 655                 getLowLevelSession();
 656
 657                 session.getNodesByIds(ids); // batch fetch
 658
 659                 Map<Serializable, String> titles = new HashMap<Serializable, String>();
 660                 for (Serializable id : ids) {
 661                         Node node = session.getNodeById(id);
 662                         if (asyncIds.contains(id)) {
 663                                 node.setSimpleProperty(Model.FULLTEXT_JOBID_PROP, id);
 664                         }
 665                         SimpleProperty prop;
 666                         try {
 667                                 prop = node.getSimpleProperty(DC_TITLE);
 668                         } catch (IllegalArgumentException e) {
 669                                 continue;
 670                         }
 671                         String title = (String) prop.getValue();
 672                         titles.put(id, title);
 673                         prop.setValue(title + " ");
 674                 }
 675                 session.save();
 676
 677                 for (Serializable id : ids) {
 678                         Node node = session.getNodeById(id);
 679                         SimpleProperty prop;
 680                         try {
 681                                 prop = node.getSimpleProperty(DC_TITLE);
 682                         } catch (IllegalArgumentException e) {
 683                                 continue;
 684                         }
 685                         prop.setValue(titles.get(id));
 686                 }
 687                 session.save();
 688         }
 689
 690         protected void runAsyncBatch(Set<String> asyncIds)
 691                         {
 692                 if (asyncIds.isEmpty()) {
 693                         return;
 694                 }
 695                 String repositoryName = coreSession.getRepositoryName();
 696                 WorkManager workManager = Framework.getLocalService(WorkManager.class);
 697                 for (String id : asyncIds) {
 698                         Work work = new SQLFulltextExtractorWork(repositoryName, id);
 699                         // schedule immediately, we're outside a transaction
 700                         workManager.schedule(work, Scheduling.IF_NOT_SCHEDULED, false);
 701                 }
 702         }
 703 }