cdmlib-io/src/main/java/eu/etaxonomy/cdm/io/common/TaxonNodeOutStreamPartitioner.java

   1 /**
   2 * Copyright (C) 2007 EDIT
   3 * European Distributed Institute of Taxonomy
   4 * http://www.e-taxonomy.eu
   5 *
   6 * The contents of this file are subject to the Mozilla Public License Version 1.1
   7 * See LICENSE.TXT at the top of this package for the full license terms.
   8 */
   9
  10 package eu.etaxonomy.cdm.io.common;
  11
  12 import java.util.ArrayList;
  13 import java.util.Arrays;
  14 import java.util.HashMap;
  15 import java.util.Iterator;
  16 import java.util.LinkedList;
  17 import java.util.List;
  18 import java.util.Map;
  19
  20 import org.apache.log4j.Logger;
  21 import org.springframework.transaction.TransactionStatus;
  22
  23 import eu.etaxonomy.cdm.api.application.ICdmRepository;
  24 import eu.etaxonomy.cdm.common.monitor.IProgressMonitor;
  25 import eu.etaxonomy.cdm.common.monitor.SubProgressMonitor;
  26 import eu.etaxonomy.cdm.filter.TaxonNodeFilter;
  27 import eu.etaxonomy.cdm.model.taxon.TaxonNode;
  28
  29 /**
  30  * @author a.mueller
  31  * @since 01.07.2017
  32  */
  33 public class TaxonNodeOutStreamPartitioner<STATE extends IoStateBase>
  34         implements ITaxonNodeOutStreamPartitioner {
  35
  36     @SuppressWarnings("unused")
  37     private static final Logger logger = Logger.getLogger(TaxonNodeOutStreamPartitioner.class);
  38
  39
  40     private static final List<String> defaultPropertyPaths = Arrays.asList(new String[]{"taxon","taxon.name"});
  41
  42     public static final List<String> fullPropertyPaths = Arrays.asList(new String[]{
  43             "*",
  44             "excludedNote.*",
  45             "classification.*",
  46             "classification.name.*",
  47             "classification.description.*",
  48             "classification.rootNode.*",
  49             "classification.rootNode.excludedNote.*",
  50             "parent.*",
  51             "agentRelations.*",
  52             "agentRelations.agent.*",
  53             "agentRelations.agent.sources.*",
  54             "agentRelations.agent.sources.citation.*",
  55             "agentRelations.agent.contact.*",
  56             "agentRelations.agent.institutionalMemberships.*",
  57             "agentRelations.agent.institutionalMemberships.institute.*",
  58             "agentRelations.agent.institutionalMemberships.institute.contact.*",
  59             "agentRelations.type.*",
  60             "agentRelations.type.representations.*",
  61             "taxon.*",
  62             "taxon.extensions.type.*",
  63             "taxon.extensions.type.representations.*",
  64             "taxon.extensions.type.vocabulary.*",
  65             "taxon.extensions.type.vocabulary.representations.*",
  66             "taxon.extensions.type.vocabulary.termRelations.*",
  67             "taxon.sources.*",
  68
  69             "taxon.name.*",
  70             "taxon.name.relationsFromThisName.*",
  71             "taxon.name.relationsToThisName.*",
  72             "taxon.name.sources.*",
  73             "taxon.name.extensions.type.*",
  74             "taxon.name.extensions.type.representations.*",
  75             "taxon.name.extensions.type.vocabulary.*",
  76             "taxon.name.extensions.type.vocabulary.terms.*",
  77             "taxon.name.extensions.type.vocabulary.terms.type.*",
  78             "taxon.name.extensions.type.vocabulary.terms.representations.*",
  79             "taxon.name.homotypicalGroup.*",
  80
  81             "taxon.synonyms.*",
  82             "taxon.synonyms.name.*",
  83             "taxon.synonyms.name.relationsFromThisName.*",
  84             "taxon.synonyms.name.relationsToThisName.*",
  85             "taxon.synonyms.name.sources.*",
  86             "taxon.synonyms.markers.type.*",
  87             "taxon.synonyms.markers.type.representations.*",
  88             "taxon.synonyms.markers.type.vocabulary.*",
  89             "taxon.synonyms.markers.type.vocabulary.terms.*",
  90             "taxon.synonyms.markers.type.vocabulary.terms.type.*",
  91             "taxon.synonyms.markers.type.vocabulary.terms.representations.*",
  92
  93
  94             "taxon.name.combinationAuthorship.*",
  95             "taxon.name.combinationAuthorship.sources.*",
  96             "taxon.name.combinationAuthorship.contact.*",
  97             "taxon.name.combinationAuthorship.teamMembers.*",
  98             "taxon.name.combinationAuthorship.teamMembers.contact.*",
  99             "taxon.name.combinationAuthorship.teamMembers.sources.*",
 100
 101             "taxon.name.exCombinationAuthorship.*",
 102             "taxon.name.basionymAuthorship.*",
 103             "taxon.name.basionymAuthorship.sources.*",
 104             "taxon.name.basionymAuthorship.contact.*",
 105             "taxon.name.basionymAuthorship.teamMembers.*",
 106             "taxon.name.basionymAuthorship.teamMembers.contact.*",
 107             "taxon.name.basionymAuthorship.teamMembers.sources.*",
 108             "taxon.name.exBasionymAuthorship.*",
 109
 110             "taxon.descriptions.*",
 111             "taxon.descriptions.elements",
 112             "taxon.descriptions.elements.*",
 113             "taxon.descriptions.elements.modifyingText.*",
 114             "taxon.descriptions.elements.sources.*",
 115             "taxon.descriptions.elements.sources.citation.*",
 116             "taxon.descriptions.elements.area.*",
 117             "taxon.descriptions.elements.area.representations.*",
 118             "taxon.descriptions.elements.area.annotations.*",
 119             "taxon.descriptions.elements.area.vocabulary.*",
 120             "taxon.descriptions.elements.area.vocabulary.terms.*",
 121             "taxon.descriptions.elements.area.vocabulary.terms.type.*",
 122             "taxon.descriptions.elements.area.vocabulary.terms.annotations.*",
 123             "taxon.descriptions.elements.area.vocabulary.terms.representations.*",
 124 //            "taxon.descriptions.elements.area.vocabulary.terms.representations.annotations.*",
 125             "taxon.descriptions.elements.area.vocabulary.representations.*",
 126
 127     });
 128
 129 //************************* STATIC ***************************************************/
 130
 131         public static <ST  extends IoStateBase>  TaxonNodeOutStreamPartitioner NewInstance(
 132                 ICdmRepository repository, ST state,
 133             TaxonNodeFilter filter, Integer partitionSize,
 134             IProgressMonitor parentMonitor, Integer parentTicks){
 135
 136             TaxonNodeOutStreamPartitioner<ST> taxonNodePartitioner
 137                         = new TaxonNodeOutStreamPartitioner(repository, state, filter, partitionSize,
 138                                 parentMonitor, parentTicks, null);
 139                 return taxonNodePartitioner;
 140         }
 141
 142     public static <ST  extends IoStateBase>  TaxonNodeOutStreamPartitioner NewInstance(
 143             ICdmRepository repository, ST state,
 144             TaxonNodeFilter filter, Integer partitionSize,
 145             IProgressMonitor parentMonitor, Integer parentTicks,List<String> propertyPath){
 146
 147         TaxonNodeOutStreamPartitioner<ST> taxonNodePartitioner
 148                 = new TaxonNodeOutStreamPartitioner(repository, state, filter, partitionSize,
 149                         parentMonitor, parentTicks, propertyPath);
 150         return taxonNodePartitioner;
 151     }
 152
 153 //*********************** VARIABLES *************************************************/
 154
 155
 156         /**
 157          * counter for the partitions
 158          */
 159         private int currentPartition;
 160
 161         private TransactionStatus txStatus;
 162
 163     private boolean readOnly = true;
 164
 165     /**
 166      * If <code>true</code> the final commit/rollback is executed only by calling
 167      * {@link #close()}
 168      */
 169     private boolean lastCommitManually = false;
 170
 171
 172
 173     private List<String> propertyPaths = defaultPropertyPaths;
 174
 175 //******************
 176
 177
 178     private final ICdmRepository repository;
 179
 180         private final TaxonNodeFilter filter;
 181
 182         private STATE state;
 183
 184     /**
 185      * Number of records handled in the partition
 186      */
 187         private final int partitionSize;
 188
 189         private int totalCount = -1;
 190         private List<Integer> idList;
 191
 192         private IProgressMonitor parentMonitor;
 193         private Integer parentTicks;
 194
 195         private SubProgressMonitor monitor;
 196
 197         private LinkedList<TaxonNode> fifo = new LinkedList<>();
 198
 199         private Iterator<Integer> idIterator;
 200
 201         private int currentIndex;
 202
 203         private static final int retrieveFactor = 1;
 204         private static final int iterateFactor = 2;
 205
 206
 207         //*********************** CONSTRUCTOR *************************************************/
 208
 209         private TaxonNodeOutStreamPartitioner(ICdmRepository repository, STATE state,
 210                 TaxonNodeFilter filter, Integer partitionSize,
 211                 IProgressMonitor parentMonitor, Integer parentTicks, List<String> propertyPaths){
 212                 this.repository = repository;
 213                 this.filter = filter;
 214                 this.partitionSize = partitionSize;
 215                 this.state = state;
 216                 this.parentMonitor = parentMonitor;
 217                 this.parentTicks = parentTicks;
 218                 if (propertyPaths != null){
 219                     this.propertyPaths = propertyPaths;
 220                 }
 221         }
 222
 223 //************************ METHODS ****************************************************/
 224
 225     public void initialize(){
 226             if (totalCount < 0){
 227
 228                 parentMonitor.subTask("Compute total number of records");
 229                 totalCount = ((Long)repository.getTaxonNodeService().count(filter)).intValue();
 230                 idList = repository.getTaxonNodeService().idList(filter);
 231                 int parTicks = this.parentTicks == null? totalCount : this.parentTicks;
 232
 233                 monitor = SubProgressMonitor.NewStarted(parentMonitor, parTicks,
 234                         "Taxon node streamer", totalCount * (retrieveFactor +  iterateFactor));
 235                 idIterator = idList.iterator();
 236                 monitor.subTask("id iterator created");
 237             }
 238         }
 239
 240
 241         @Override
 242     public TaxonNode next(){
 243             int currentIndexAtStart = currentIndex;
 244             initialize();
 245             if(fifo.isEmpty()){
 246                 List<TaxonNode> list = getNextPartition();
 247                 fifo.addAll(list);
 248             }
 249             if (!fifo.isEmpty()){
 250                 TaxonNode result = fifo.removeFirst();
 251                 // worked should be called after each step is ready,
 252                 //this is usually after each next() call but not for the first
 253                 if (currentIndexAtStart > 0){
 254                     monitor.worked(iterateFactor);
 255                 }
 256                 return result;
 257             }else{
 258                 if(!lastCommitManually){
 259                     commitTransaction();
 260                 }
 261                 return null;
 262             }
 263         }
 264
 265         @Override
 266     public void close(){
 267             monitor.done();
 268             commitTransaction();
 269         }
 270
 271     private List<TaxonNode> getNextPartition() {
 272         List<Integer> partList = new ArrayList<>();
 273
 274         if (txStatus != null){
 275             commitTransaction();
 276         }
 277
 278         txStatus = startTransaction();
 279 //        if (readOnly){
 280 //            txStatus.setRollbackOnly();
 281 //        }
 282         while (partList.size() < partitionSize && idIterator.hasNext()){
 283             partList.add(idIterator.next());
 284             currentIndex++;
 285         }
 286         List<TaxonNode> partition = new ArrayList<>();
 287         if (!partList.isEmpty()){
 288             monitor.subTask(String.format("Reading partition %d/%d", currentPartition + 1, (totalCount / partitionSize) +1 ));
 289             partition = repository.getTaxonNodeService().loadByIds(partList, propertyPaths);
 290             monitor.worked(partition.size());
 291             currentPartition++;
 292             monitor.subTask(String.format("Writing partition %d/%d", currentPartition, (totalCount / partitionSize) +1 ));
 293         }
 294         return partition;
 295     }
 296
 297     private void commitTransaction() {
 298         if (!txStatus.isCompleted()){
 299             if (this.readOnly){
 300                 repository.rollbackTransaction(txStatus);
 301             }else{
 302                 repository.commitTransaction(txStatus);
 303             }
 304         }
 305     }
 306
 307     private TransactionStatus startTransaction() {
 308         return repository.startTransaction(readOnly);
 309     }
 310
 311     @Override
 312     public void setReadOnly(boolean readOnly) {
 313         this.readOnly = readOnly;
 314     }
 315
 316     public boolean isLastCommitManually() {
 317         return lastCommitManually;
 318     }
 319
 320     public void setLastCommitManually(boolean lastCommitManually) {
 321         this.lastCommitManually = lastCommitManually;
 322     }
 323 }