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

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