bundles/org.simantics.db.common/src/org/simantics/db/common/processor/MergingWriteOnlyProcessor.java

   1 /*******************************************************************************
   2  * Copyright (c) 2007, 2010 Association for Decentralized Information Management
   3  * in Industry THTH ry.
   4  * All rights reserved. This program and the accompanying materials
   5  * are made available under the terms of the Eclipse Public License v1.0
   6  * which accompanies this distribution, and is available at
   7  * http://www.eclipse.org/legal/epl-v10.html
   8  *
   9  * Contributors:
  10  *     VTT Technical Research Centre of Finland - initial API and implementation
  11  *******************************************************************************/
  12 package org.simantics.db.common.processor;
  13
  14 import java.util.concurrent.ConcurrentLinkedQueue;
  15
  16 import org.simantics.db.AsyncRequestProcessor;
  17 import org.simantics.db.VirtualGraph;
  18 import org.simantics.db.WriteOnlyGraph;
  19 import org.simantics.db.common.request.WriteOnlyRequest;
  20 import org.simantics.db.common.utils.Logger;
  21 import org.simantics.db.exception.DatabaseException;
  22 import org.simantics.db.request.WriteOnly;
  23
  24 final public class MergingWriteOnlyProcessor extends ProcessorBase {
  25
  26     final long transactionKeepalivePeriod;
  27     final ConcurrentLinkedQueue<WriteOnly> queue = new ConcurrentLinkedQueue<WriteOnly>();
  28     final private AsyncRequestProcessor processor;
  29     final private VirtualGraph vg;
  30
  31     boolean hasAlreadyRequest = false;
  32
  33     public MergingWriteOnlyProcessor(AsyncRequestProcessor processor, VirtualGraph vg, long transactionKeepalivePeriod) {
  34         this.processor = processor;
  35         this.vg = vg;
  36         this.transactionKeepalivePeriod = transactionKeepalivePeriod;
  37     }
  38
  39     class MergedWrite extends WriteOnlyRequest {
  40
  41         public MergedWrite(VirtualGraph vg) {
  42                 super(vg);
  43                 }
  44
  45         @Override
  46         public void perform(WriteOnlyGraph graph) throws DatabaseException {
  47
  48 //            System.out.println("MergedWrite begins");
  49
  50             while (true) {
  51
  52                 WriteOnly next = queue.poll();
  53                 if(next == null) {
  54                     synchronized (MergingWriteOnlyProcessor.this) {
  55                         if (transactionKeepalivePeriod > 0) {
  56                             try {
  57                                 MergingWriteOnlyProcessor.this.wait(transactionKeepalivePeriod);
  58                             } catch (InterruptedException e) {
  59                                         Logger.defaultLogError(e);
  60                             }
  61                             next = queue.poll();
  62                         }
  63                         if(next == null) {
  64                             hasAlreadyRequest = false;
  65 //                            System.out.println("MergedWrite ends");
  66                             return;
  67                         }
  68                     }
  69                 }
  70
  71 //                System.out.println("MergedWrite executes " + next);
  72                 try {
  73                         next.perform(graph);
  74                 } catch(Throwable t) {
  75                         Logger.defaultLogError(t);
  76                 }
  77
  78             }
  79
  80         }
  81
  82     }
  83
  84     @Override
  85     public void asyncRequest(WriteOnly request) {
  86
  87         queue.add(request);
  88
  89         synchronized (this) {
  90
  91           if (!hasAlreadyRequest) {
  92               processor.asyncRequest(new MergedWrite(vg));
  93               hasAlreadyRequest = true;
  94           } else {
  95               notify();
  96           }
  97
  98         }
  99
 100     }
 101
 102     @Override
 103     public String toString() {
 104         return "MergingWriteOnlyProcessor@" + System.identityHashCode(this) + " (based on " + processor + ")";
 105     }
 106
 107 }