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

   1 /*******************************************************************************
   2  * Copyright (c) 2007, 2018 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.ArrayList;
  15 import java.util.Collection;
  16 import java.util.HashMap;
  17 import java.util.HashSet;
  18 import java.util.LinkedList;
  19 import java.util.Set;
  20 import java.util.UUID;
  21 import java.util.concurrent.Semaphore;
  22 import java.util.function.Consumer;
  23
  24 import org.simantics.db.AsyncReadGraph;
  25 import org.simantics.db.AsyncRequestProcessor;
  26 import org.simantics.db.ReadGraph;
  27 import org.simantics.db.Resource;
  28 import org.simantics.db.Session;
  29 import org.simantics.db.WriteGraph;
  30 import org.simantics.db.common.procedure.adapter.AsyncMultiProcedureAdapter;
  31 import org.simantics.db.common.procedure.adapter.ProcedureAdapter;
  32 import org.simantics.db.common.procedure.adapter.SyncMultiProcedureAdapter;
  33 import org.simantics.db.common.procedure.wrapper.NoneToAsyncProcedure;
  34 import org.simantics.db.common.request.ReadRequest;
  35 import org.simantics.db.common.request.WriteRequest;
  36 import org.simantics.db.common.utils.Logger;
  37 import org.simantics.db.exception.CancelTransactionException;
  38 import org.simantics.db.exception.DatabaseException;
  39 import org.simantics.db.procedure.AsyncListener;
  40 import org.simantics.db.procedure.AsyncMultiListener;
  41 import org.simantics.db.procedure.AsyncMultiProcedure;
  42 import org.simantics.db.procedure.AsyncProcedure;
  43 import org.simantics.db.procedure.Listener;
  44 import org.simantics.db.procedure.MultiListener;
  45 import org.simantics.db.procedure.MultiProcedure;
  46 import org.simantics.db.procedure.Procedure;
  47 import org.simantics.db.procedure.SyncListener;
  48 import org.simantics.db.procedure.SyncMultiListener;
  49 import org.simantics.db.procedure.SyncMultiProcedure;
  50 import org.simantics.db.procedure.SyncProcedure;
  51 import org.simantics.db.request.AsyncMultiRead;
  52 import org.simantics.db.request.AsyncRead;
  53 import org.simantics.db.request.DelayedWrite;
  54 import org.simantics.db.request.DelayedWriteResult;
  55 import org.simantics.db.request.ExternalRead;
  56 import org.simantics.db.request.MultiRead;
  57 import org.simantics.db.request.Read;
  58 import org.simantics.db.request.ReadInterface;
  59 import org.simantics.db.request.Write;
  60 import org.simantics.db.request.WriteInterface;
  61 import org.simantics.db.request.WriteOnly;
  62 import org.simantics.db.request.WriteOnlyResult;
  63 import org.simantics.db.request.WriteResult;
  64 import org.simantics.utils.DataContainer;
  65 import org.simantics.utils.datastructures.Pair;
  66
  67 public class MergingGraphRequestProcessor implements AsyncRequestProcessor {
  68
  69     private static class SyncWriteRequestAdapter implements Write {
  70
  71         private Semaphore semaphore = new Semaphore(0);
  72         private Object request;
  73         private Throwable exception;
  74         SyncWriteRequestAdapter(Write r) {
  75             this.request = r;
  76         }
  77         SyncWriteRequestAdapter(WriteOnly r) {
  78             this.request = r;
  79         }
  80 //        @Override
  81 //        public GraphRequestStatus perform(Graph g) throws Exception {
  82 //            return perform((ReadGraph)g);
  83 //        }
  84         @Override
  85         public void perform(WriteGraph g) throws DatabaseException, CancelTransactionException {
  86             if(request instanceof Write) {
  87                 ((Write)request).perform(g);
  88             } else if(request instanceof DelayedWrite) {
  89                     ((DelayedWrite)request).perform(g);
  90             } else {
  91                 ((WriteOnly)request).perform(g);
  92             }
  93         }
  94 //        @Override
  95 //        public String getId() {
  96 //            if(request instanceof WriteGraphRequest) {
  97 //                return ((WriteGraphRequest)request).getId();
  98 //            } else {
  99 //                return null;
 100 //            }
 101 //        }
 102 //        @Override
 103 //        public void requestCompleted(GraphRequestStatus status) {
 104 //            if(request instanceof WriteGraphRequest) {
 105 //                ((WriteGraphRequest)request).requestCompleted(status);
 106 //            } else {
 107 //            }
 108 //        }
 109 //        @Override
 110 //        public void handleException(Throwable e) {
 111 //            this.exception = e;
 112 //            if(request instanceof WriteGraphRequest) {
 113 //                ((WriteGraphRequest)request).handleException(e);
 114 //            }
 115 //        }
 116
 117         public void throwOrWrapException() {
 118             if (exception == null)
 119                 return;
 120             if (exception instanceof RuntimeException)
 121                 throw (RuntimeException) exception;
 122             if (exception instanceof Error)
 123                 throw (Error) exception;
 124             throw new RuntimeException("See cause for the real exception.", exception);
 125         }
 126
 127         public void acquire() {
 128             try {
 129                 semaphore.acquire();
 130             } catch (InterruptedException e) {
 131                         Logger.defaultLogError(e);
 132             }
 133         }
 134
 135         public void release() {
 136             semaphore.release();
 137         }
 138
 139         @Override
 140         public String toString() {
 141             return "SyncWriteRequestAdapter " + request;
 142         }
 143
 144     }
 145
 146     long transactionKeepalivePeriod;
 147
 148     /**
 149      * Synchronization object for implementing {@link #synchronize()}.
 150      * {@link Object#notifyAll()} is invoked for this lock object every time a
 151      * single transaction is completed, thereby releasing all waiters in
 152      * {@link #synchronize()}.
 153      */
 154     Object barrier = new Object();
 155
 156     Set<Pair<Object, Object>> requestSet = new HashSet<Pair<Object, Object>>();
 157     LinkedList<Pair<Object, Object>> requestQueue = new LinkedList<Pair<Object, Object>>();
 158     boolean hasAlreadyRequest = false;
 159
 160     /**
 161      * A set of requests which {@link #synchronize()} is depending on at the
 162      * moment. Every time a request within this set is completed, some thread in
 163      * {@link #synchronize()} should be released.
 164      */
 165 //    Set<Object> barrierRequests = new HashSet<Object>();
 166     Set<Object> syncRequests = new HashSet<Object>();
 167
 168     private String name;
 169
 170     private AsyncRequestProcessor processor;
 171
 172     public MergingGraphRequestProcessor(String name, AsyncRequestProcessor processor, long transactionKeepalivePeriod) {
 173         this.name = name;
 174         this.processor = processor;
 175         this.transactionKeepalivePeriod = transactionKeepalivePeriod;
 176     }
 177
 178     public MergingGraphRequestProcessor(AsyncRequestProcessor processor, long transactionKeepalivePeriod) {
 179         this.name = "MergingGraphRequestProcessor" + UUID.randomUUID().toString();
 180         this.processor = processor;
 181         this.transactionKeepalivePeriod = transactionKeepalivePeriod;
 182     }
 183
 184     @SuppressWarnings({"unchecked", "rawtypes"})
 185     protected class MergedRead extends ReadRequest {
 186
 187         Pair<Object, Object> currentRequest;
 188
 189 //        RunnerReadGraphRequest(GraphRequestProcessor processor) {
 190 //            super(processor);
 191 //        }
 192 //
 193 //        @Override
 194 //        public void completed(boolean value) {
 195 ////            System.out.println(this + "MGRP read completed");
 196 ////            synchronized (MergingGraphRequestProcessor.this) {
 197 ////                if (requestQueue.isEmpty())
 198 ////                    hasAlreadyRequest = false;
 199 ////                else
 200 ////                    newTransaction();
 201 ////            }
 202 //        }
 203
 204         @Override
 205         public void run(ReadGraph graph) {
 206
 207 //            System.out.println(MergingGraphRequestProcessor.this + " reads");
 208
 209             while (true) {
 210
 211                 synchronized (MergingGraphRequestProcessor.this) {
 212
 213                     // Release #synchronize() invokers if necessary.
 214 //                    if (currentRequest != null && barrierRequests.contains(currentRequest)) {
 215 //                        synchronized (barrier) {
 216 //                            barrier.notifyAll();
 217 //                        }
 218 //                    }
 219
 220                     if(requestQueue.isEmpty()) {
 221                         if (transactionKeepalivePeriod > 0) {
 222 //                            System.out.println("MGRP [" + MergingGraphRequestProcessor.this + "] waits " + transactionKeepalivePeriod + " ms. in " + Thread.currentThread() );
 223                             try {
 224                                 MergingGraphRequestProcessor.this.wait(transactionKeepalivePeriod);
 225                             } catch (InterruptedException e) {
 226                                         Logger.defaultLogError(e);
 227                             }
 228                             if (requestQueue.isEmpty())
 229                                 break;
 230                         } else
 231                             break;
 232                     }
 233
 234                     Object nextRequest = requestQueue.peekFirst().first;
 235                     if(nextRequest instanceof Write || nextRequest instanceof DelayedWrite) {
 236                         break;
 237                     }
 238
 239                     currentRequest = requestQueue.remove(0);
 240                     requestSet.remove(currentRequest);
 241
 242                 }
 243
 244 //                ReadGraphRequest req = (ReadGraphRequest)currentRequest.first;
 245
 246                 if( syncRequests.contains(currentRequest.first)) {
 247
 248                     try {
 249
 250                         if(currentRequest.second instanceof AsyncProcedure<?>) {
 251                             if(currentRequest.first instanceof Read) {
 252                                 Read req = (Read)currentRequest.first;
 253                                 graph.syncRequest(req, (AsyncProcedure<?>)currentRequest.second);
 254                             } else {
 255                                 AsyncRead req = (AsyncRead)currentRequest.first;
 256                                 graph.syncRequest(req, (AsyncProcedure<?>)currentRequest.second);
 257                             }
 258                         } else {
 259                             AsyncMultiRead req = (AsyncMultiRead)currentRequest.first;
 260                             graph.syncRequest(req, (AsyncMultiProcedure<?>)currentRequest.second);
 261                         }
 262
 263                     } catch(Throwable t) {
 264
 265                                 Logger.defaultLogError(t);
 266
 267 //                        if(currentRequest.second instanceof AsyncProcedure<?>) {
 268 //                            ((AsyncProcedure<?>)currentRequest.second).exception(graph, t);
 269 //                        } else {
 270 //                            ((AsyncMultiProcedure<?>)currentRequest.second).exception(graph, t);
 271 //                        }
 272
 273                     }
 274
 275                     synchronized (currentRequest.first) {
 276                         syncRequests.remove(currentRequest.first);
 277 //                      System.out.println("notifying " + currentRequest.first);
 278                         currentRequest.first.notify();
 279                     }
 280
 281
 282                 } else {
 283
 284                     try{
 285
 286                         if(currentRequest.second instanceof AsyncProcedure<?>) {
 287                             if(currentRequest.first instanceof AsyncRead) {
 288                                 AsyncRead req = (AsyncRead)currentRequest.first;
 289                                 graph.syncRequest(req, (AsyncProcedure<?>)currentRequest.second);
 290                             } else {
 291                                 Read req = (Read)currentRequest.first;
 292                                 graph.syncRequest(req, (AsyncProcedure<?>)currentRequest.second);
 293                             }
 294                         } else {
 295                             AsyncMultiRead req = (AsyncMultiRead)currentRequest.first;
 296                             graph.syncRequest(req, (AsyncMultiProcedure<?>)currentRequest.second);
 297                         }
 298
 299                     } catch(Throwable t) {
 300
 301                                 Logger.defaultLogError(t);
 302
 303 //                        if(currentRequest.second instanceof AsyncProcedure<?>) {
 304 //                            ((AsyncProcedure<?>)currentRequest.second).exception(graph, t);
 305 //                        } else {
 306 //                            ((AsyncMultiProcedure<?>)currentRequest.second).exception(graph, t);
 307 //                        }
 308
 309                     }
 310                 }
 311
 312             }
 313
 314 //            System.out.println(MergingGraphRequestProcessor.this + " read completed");
 315
 316             synchronized (MergingGraphRequestProcessor.this) {
 317                 if (requestQueue.isEmpty())
 318                     hasAlreadyRequest = false;
 319                 else
 320                     newTransaction();
 321             }
 322
 323         }
 324
 325         @Override
 326         public String toString() {
 327             return "MergedRead[" + requestQueue.size() + " requests]";
 328         }
 329
 330     }
 331
 332     protected class RunnerWriteGraphRequest extends WriteRequest {
 333
 334         Pair<Object, Object> currentRequest;
 335         HashMap<String, String> metadata = new HashMap<String, String>();
 336
 337         @Override
 338         public void perform(WriteGraph graph) throws DatabaseException {
 339
 340 //            System.out.println(MergingGraphRequestProcessor.this + " writes");
 341
 342             while (true) {
 343
 344                 synchronized (MergingGraphRequestProcessor.this) {
 345
 346                     // Release #synchronize() invokers if necessary.
 347 //                    if (currentRequest != null && barrierRequests.contains(currentRequest)) {
 348 //                        synchronized (barrier) {
 349 //                            barrier.notifyAll();
 350 //                        }
 351 //                    }
 352
 353                     if(requestQueue.isEmpty()) {
 354                         if (transactionKeepalivePeriod > 0) {
 355                             try {
 356                                 MergingGraphRequestProcessor.this.wait(transactionKeepalivePeriod);
 357                             } catch (InterruptedException e) {
 358                                         Logger.defaultLogError(e);
 359                             }
 360                             if (requestQueue.isEmpty())
 361                                 break;
 362                         } else
 363                             break;
 364                     }
 365
 366                     Object nextRequest = requestQueue.peekFirst().first;
 367                     if(nextRequest instanceof AsyncMultiRead || nextRequest instanceof AsyncRead || nextRequest instanceof Read) {
 368                         break;
 369                     }
 370
 371                     currentRequest = requestQueue.remove(0);
 372                     requestSet.remove(currentRequest);
 373
 374                 }
 375
 376                 @SuppressWarnings("unchecked")
 377                 Consumer<Throwable> callback = (Consumer<Throwable>)currentRequest.second;
 378
 379                 if (currentRequest.first.getClass().equals(SyncWriteRequestAdapter.class)) {
 380
 381                     SyncWriteRequestAdapter adapter = (SyncWriteRequestAdapter)currentRequest.first;
 382
 383                     try {
 384 //                        System.out.println("merg.sync " + adapter);
 385                         graph.syncRequest(adapter);
 386                         if(callback != null) callback.accept(null);
 387                     } catch(Throwable t) {
 388                                 Logger.defaultLogError(t);
 389                         if(callback != null) callback.accept(t);
 390                     }
 391
 392                     adapter.release();
 393 //                    System.out.println("merg.sync.release " + adapter);
 394
 395                 } else {
 396
 397                     try {
 398                         if(currentRequest.first instanceof Write) graph.syncRequest((Write)currentRequest.first);
 399                         else if(currentRequest.first instanceof DelayedWrite) graph.syncRequest((DelayedWrite)currentRequest.first);
 400                         if(callback != null) callback.accept(null);
 401                     } catch(Throwable t) {
 402                                 Logger.defaultLogError(t);
 403                         if(callback != null) callback.accept(t);
 404                     }
 405
 406                 }
 407
 408             }
 409
 410 //            System.out.println(MergingGraphRequestProcessor.this + " write completed");
 411
 412             synchronized (MergingGraphRequestProcessor.this) {
 413                 if (requestQueue.isEmpty())
 414                     hasAlreadyRequest = false;
 415                 else
 416                     newTransaction();
 417             }
 418
 419         }
 420
 421     }
 422
 423     private void newTransaction() {
 424
 425         boolean write = false;
 426
 427         synchronized (MergingGraphRequestProcessor.this) {
 428             assert(!requestQueue.isEmpty());
 429             Object nextRequest = requestQueue.peekFirst().first;
 430             write = (nextRequest instanceof Write || nextRequest instanceof DelayedWrite);
 431         }
 432
 433         if(write) {
 434             processor.asyncRequest(new RunnerWriteGraphRequest(), null);
 435         } else {
 436             processor.asyncRequest(new MergedRead());
 437         }
 438
 439     }
 440
 441     @Override
 442     public <T> void asyncRequest(AsyncMultiRead<T> request, AsyncMultiProcedure<T> procedure) {
 443
 444 //        System.out.println(this + " asyncRequest(ReadGraphRequest<QueryProcedure4<T>> request, QueryProcedure4<T> procedure)");
 445
 446         Pair<Object, Object> pair = Pair.make(request, procedure);
 447         if (requestSet.contains(pair))
 448             return;
 449
 450         requestQueue.add(pair);
 451         requestSet.add(pair);
 452
 453         if (!hasAlreadyRequest) {
 454             newTransaction();
 455             hasAlreadyRequest = true;
 456         } else {
 457             notify();
 458         }
 459
 460     }
 461
 462     @Override
 463     public synchronized <T> void asyncRequest(AsyncRead<T> request, AsyncProcedure<T> procedure) {
 464
 465 //        System.out.println(this + " asyncRequest(ReadGraphRequest<SingleQueryProcedure4<T>> request, SingleQueryProcedure4<T> procedure) " + this);
 466
 467         Pair<Object, Object> pair = Pair.make(request, procedure);
 468         if (requestSet.contains(pair))
 469             return;
 470
 471         requestQueue.add(pair);
 472         requestSet.add(pair);
 473
 474         if (!hasAlreadyRequest) {
 475             newTransaction();
 476             hasAlreadyRequest = true;
 477         } else {
 478 //            System.out.println("notify " + this);
 479             notify();
 480         }
 481
 482     }
 483
 484     @Override
 485     public synchronized void asyncRequest(Write request, Consumer<DatabaseException> callback) {
 486
 487 //        System.out.println(this + " asyncRequest(WriteGraphRequest request)");
 488
 489         Pair<Object, Object> pair = Pair.make(request, callback);
 490         if (requestSet.contains(pair))
 491             return;
 492
 493         requestQueue.add(pair);
 494         requestSet.add(pair);
 495
 496         if (!hasAlreadyRequest) {
 497 //            System.out.println("new transaction");
 498             newTransaction();
 499             hasAlreadyRequest = true;
 500         } else {
 501 //            System.out.println("notify");
 502             notify();
 503         }
 504
 505     }
 506
 507     @Override
 508     public synchronized void asyncRequest(DelayedWrite request, Consumer<DatabaseException> callback) {
 509
 510 //        System.out.println(this + " asyncRequest(WriteGraphRequest request)");
 511
 512         Pair<Object, Object> pair = Pair.make(request, callback);
 513         if (requestSet.contains(pair))
 514             return;
 515
 516         requestQueue.add(pair);
 517         requestSet.add(pair);
 518
 519         if (!hasAlreadyRequest) {
 520 //            System.out.println("new transaction");
 521             newTransaction();
 522             hasAlreadyRequest = true;
 523         } else {
 524 //            System.out.println("notify");
 525             notify();
 526         }
 527
 528     }
 529
 530     @Override
 531     public synchronized void asyncRequest(WriteOnly request, Consumer<DatabaseException> callback) {
 532
 533 //        System.out.println(this + " asyncRequest(WriteGraphRequest request)");
 534
 535         Pair<Object, Object> pair = Pair.make(request, callback);
 536         if (requestSet.contains(pair))
 537             return;
 538
 539         requestQueue.add(pair);
 540         requestSet.add(pair);
 541
 542         if (!hasAlreadyRequest) {
 543 //            System.out.println("new transaction");
 544             newTransaction();
 545             hasAlreadyRequest = true;
 546         } else {
 547 //            System.out.println("notify");
 548             notify();
 549         }
 550
 551     }
 552
 553     @Override
 554     public <T> Collection<T> syncRequest(AsyncMultiRead<T> request, final AsyncMultiProcedure<T> procedure) {
 555
 556         final DataContainer<Throwable> throwable = new DataContainer<Throwable>(null);
 557
 558         // Queue the adapter up for execution.
 559         synchronized (request) {
 560             syncRequests.add(request);
 561             asyncRequest(request, procedure);
 562             if(syncRequests.contains(request)) {
 563                 try {
 564                     //                  System.out.println("waiting " + request);
 565                     request.wait();
 566                 } catch (InterruptedException e) {
 567                     throw new Error(e);
 568                 }
 569             }
 570         }
 571
 572         Throwable t = throwable.get();
 573
 574         if(t != null) {
 575                 Logger.defaultLogError(t);
 576             throw new RuntimeException(t.getMessage());
 577         }
 578
 579         return null;
 580
 581     }
 582
 583     @Override
 584     public <T> T syncRequest(AsyncRead<T> request, final AsyncProcedure<T> procedure) {
 585
 586 //        System.out.println("syncRequest(ReadGraphRequest<SingleQueryProcedure4<T>> request, SingleQueryProcedure4<T> procedure)");
 587
 588         final DataContainer<T> result = new DataContainer<T>(null);
 589         final DataContainer<Throwable> throwable = new DataContainer<Throwable>(null);
 590
 591         // Queue the adapter up for execution.
 592         synchronized (request) {
 593
 594             syncRequests.add(request);
 595             asyncRequest(request, new AsyncProcedure<T>() {
 596
 597                 public void execute(AsyncReadGraph graph, T t) {
 598                     synchronized(result) {
 599                         result.set(t);
 600                     }
 601                     procedure.execute(graph, t);
 602                 };
 603
 604                 @Override
 605                 public void exception(AsyncReadGraph graph, Throwable t) {
 606                     throwable.set(t);
 607                 }
 608
 609                 @Override
 610                 public String toString() {
 611                     return procedure.toString();
 612                 }
 613
 614             });
 615             if(syncRequests.contains(request)) {
 616                 try {
 617                     //                  System.out.println("waiting " + request);
 618                     request.wait();
 619                 } catch (InterruptedException e) {
 620                     throw new Error(e);
 621                 }
 622             }
 623         }
 624
 625         Throwable t = throwable.get();
 626
 627         if(t != null) {
 628                 Logger.defaultLogError(t);
 629             throw new RuntimeException(t.getMessage());
 630         }
 631
 632         return result.get();
 633
 634         //return result.get();
 635
 636     }
 637
 638
 639     @Override
 640     public void syncRequest(Write request) {
 641
 642 //        System.out.println(MergingGraphRequestProcessor.this + " syncRequest(WriteGraphRequest)");
 643
 644         SyncWriteRequestAdapter adapter = new SyncWriteRequestAdapter(request);
 645
 646         asyncRequest(adapter, null);
 647
 648         adapter.acquire();
 649
 650         // Throw exception if one occurred.
 651         adapter.throwOrWrapException();
 652
 653     }
 654
 655     @Override
 656     public void syncRequest(WriteOnly request) {
 657
 658 //        System.out.println(MergingGraphRequestProcessor.this + " syncRequest(WriteGraphRequest)");
 659
 660         SyncWriteRequestAdapter adapter = new SyncWriteRequestAdapter(request);
 661
 662         // Queue the adapter up for execution.
 663         synchronized (adapter) {
 664             asyncRequest(adapter, null);
 665             try {
 666                 adapter.wait();
 667             } catch (InterruptedException e) {
 668                 throw new Error(e);
 669             }
 670         }
 671
 672         // Throw exception if one occurred.
 673         adapter.throwOrWrapException();
 674
 675     }
 676
 677     @Override
 678     public Session getSession() {
 679         return processor.getSession();
 680     }
 681
 682     @Override
 683     public String toString() {
 684         return "MergingGraphRequestProcessor[" + name + "]@" + System.identityHashCode(this) + " (based on " + processor + ")";
 685     }
 686
 687     @Override
 688     public <T> void asyncRequest(AsyncRead<T> request) {
 689
 690         asyncRequest(request, new ProcedureAdapter<T>() {
 691
 692             @Override
 693             public void exception(Throwable t) {
 694                         Logger.defaultLogError(t);
 695             }
 696
 697         });
 698
 699     }
 700
 701     @Override
 702     public <T> void asyncRequest(AsyncRead<T> request, Procedure<T> procedure) {
 703         asyncRequest(request, new NoneToAsyncProcedure<T>(procedure));
 704     }
 705
 706     @Override
 707     public <T> void asyncRequest(AsyncMultiRead<T> request) {
 708         throw new UnsupportedOperationException("Not implemented.");
 709     }
 710
 711     @Override
 712     public <T> void asyncRequest(AsyncMultiRead<T> request,
 713             MultiProcedure<T> procedure) {
 714         throw new UnsupportedOperationException("Not implemented.");
 715     }
 716
 717     @Override
 718     public <T> T syncRequest(AsyncRead<T> request) {
 719
 720         final DataContainer<Throwable> throwable = new DataContainer<Throwable>(null);
 721         final DataContainer<T> result = new DataContainer<T>();
 722
 723         syncRequest(request, new AsyncProcedure<T>() {
 724
 725             public void execute(AsyncReadGraph graph, T t) {
 726                 result.set(t);
 727             }
 728
 729             @Override
 730             public void exception(AsyncReadGraph graph, Throwable t) {
 731                 throwable.set(t);
 732             }
 733
 734         });
 735
 736         Throwable t = throwable.get();
 737
 738         if(t != null) {
 739                 Logger.defaultLogError(t);
 740             throw new RuntimeException(t.getMessage());
 741         }
 742
 743         return result.get();
 744
 745     }
 746
 747     @Override
 748     public <T> T syncRequest(AsyncRead<T> request,
 749             Procedure<T> procedure) {
 750         throw new UnsupportedOperationException("Not implemented.");
 751     }
 752
 753     @Override
 754     public <T> Collection<T> syncRequest(AsyncMultiRead<T> request) {
 755
 756         final DataContainer<Throwable> throwable = new DataContainer<Throwable>(null);
 757         final ArrayList<T> result = new ArrayList<T>();
 758
 759         syncRequest(request, new AsyncMultiProcedureAdapter<T>() {
 760
 761             @Override
 762             public void execute(AsyncReadGraph graph, T t) {
 763                 synchronized(result) {
 764                     result.add(t);
 765                 }
 766             };
 767
 768             @Override
 769             public void exception(AsyncReadGraph graph, Throwable t) {
 770                 throwable.set(t);
 771             }
 772
 773         });
 774
 775         Throwable t = throwable.get();
 776
 777         if(t != null) {
 778                 Logger.defaultLogError(t);
 779             throw new RuntimeException(t.getMessage());
 780         }
 781
 782         return result;
 783
 784     }
 785
 786     @Override
 787     public <T> Collection<T> syncRequest(AsyncMultiRead<T> request,
 788             MultiProcedure<T> procedure) {
 789         throw new Error("Not implemented.");
 790     }
 791
 792     @Override
 793     public <T> T syncRequest(Read<T> request) {
 794
 795         final DataContainer<Throwable> throwable = new DataContainer<Throwable>(null);
 796         final DataContainer<T> result = new DataContainer<T>();
 797
 798
 799         syncRequest(request, new Procedure<T>() {
 800
 801             public void execute(T t) {
 802                 result.set(t);
 803             }
 804
 805             @Override
 806             public void exception(Throwable t) {
 807                 throwable.set(t);
 808             }
 809
 810         });
 811
 812         Throwable t = throwable.get();
 813
 814         if(t != null) {
 815             throw new Error(t.getMessage());
 816         }
 817
 818
 819         return result.get();
 820
 821     }
 822
 823     @Override
 824     public <T> T syncRequest(Read<T> request,
 825             final AsyncProcedure<T> procedure) {
 826
 827         final DataContainer<T> result = new DataContainer<T>(null);
 828         final DataContainer<Throwable> throwable = new DataContainer<Throwable>(null);
 829
 830         // Queue the adapter up for execution.
 831         synchronized (request) {
 832
 833             syncRequests.add(request);
 834             asyncRequest(request, new AsyncProcedure<T>() {
 835
 836                 public void execute(AsyncReadGraph graph, T t) {
 837                     synchronized(result) {
 838                         result.set(t);
 839                     }
 840                     procedure.execute(graph, t);
 841                 };
 842
 843                 @Override
 844                 public void exception(AsyncReadGraph graph, Throwable t) {
 845                     throwable.set(t);
 846                 }
 847
 848                 @Override
 849                 public String toString() {
 850                     return procedure.toString();
 851                 }
 852
 853             });
 854             if(syncRequests.contains(request)) {
 855                 try {
 856                     //                  System.out.println("waiting " + request);
 857                     request.wait();
 858                 } catch (InterruptedException e) {
 859                     throw new Error(e);
 860                 }
 861             }
 862         }
 863
 864         Throwable t = throwable.get();
 865
 866         if(t != null) {
 867             throw new RuntimeException("Unexpected exception in MergingGraphRequestProcessor.syncRequest(Read, AsyncProcedure)", t);
 868         }
 869
 870         return result.get();
 871
 872     }
 873
 874     @Override
 875     public <T> void asyncRequest(Read<T> request) {
 876
 877         asyncRequest(request, new ProcedureAdapter<T>() {
 878
 879             @Override
 880             public void exception(Throwable t) {
 881                 Logger.defaultLogError(t);
 882             }
 883
 884         });
 885
 886     }
 887
 888     @Override
 889     public synchronized <T> void asyncRequest(Read<T> request,
 890             AsyncProcedure<T> procedure) {
 891
 892         Pair<Object, Object> pair = Pair.make(request, procedure);
 893         if (requestSet.contains(pair))
 894             return;
 895
 896         requestQueue.add(pair);
 897         requestSet.add(pair);
 898
 899         if (!hasAlreadyRequest) {
 900             newTransaction();
 901             hasAlreadyRequest = true;
 902         } else {
 903             notify();
 904         }
 905
 906     }
 907
 908     @Override
 909     public <T> T syncRequest(Read<T> request, Procedure<T> procedure) {
 910         return syncRequest(request, new NoneToAsyncProcedure<T>(procedure));
 911     }
 912
 913     @Override
 914     public <T> Collection<T> syncRequest(final MultiRead<T> request) throws DatabaseException {
 915         assert(request != null);
 916
 917         final ArrayList<T> result = new ArrayList<T>();
 918         final DataContainer<Throwable> exception = new DataContainer<Throwable>();
 919
 920         syncRequest(request, new SyncMultiProcedureAdapter<T>() {
 921
 922             @Override
 923             public void execute(ReadGraph graph, T t) {
 924                 synchronized(result) {
 925                     result.add(t);
 926                 }
 927             }
 928
 929             @Override
 930             public void exception(ReadGraph graph, Throwable t) {
 931                 exception.set(t);
 932             }
 933
 934             @Override
 935             public String toString() {
 936                 return "syncRequest(MultiRead) -> " + request;
 937             }
 938
 939         });
 940
 941         Throwable t = exception.get();
 942         if(t != null) {
 943             if(t instanceof DatabaseException) throw (DatabaseException)t;
 944             else throw new DatabaseException("Unexpected exception in ReadGraph.syncRequest(Read)", t);
 945         }
 946
 947         return result;
 948     }
 949
 950     @Override
 951     public <T> Collection<T> syncRequest(MultiRead<T> request, MultiProcedure<T> procedure) {
 952         throw new UnsupportedOperationException("Not implemented");
 953     }
 954
 955     @Override
 956     public <T> void asyncRequest(Read<T> request, Procedure<T> procedure) {
 957         asyncRequest(request, new NoneToAsyncProcedure<T>(procedure));
 958     }
 959
 960     @Override
 961     public <T> void asyncRequest(MultiRead<T> request) {
 962         throw new UnsupportedOperationException("Not implemented");
 963     }
 964
 965     @Override
 966     public <T> void asyncRequest(MultiRead<T> request, MultiProcedure<T> procedure) {
 967         throw new UnsupportedOperationException("Not implemented");
 968     }
 969
 970     @Override
 971     public void asyncRequest(Write r) {
 972         asyncRequest(r, null);
 973     }
 974
 975     @Override
 976     public void asyncRequest(DelayedWrite r) {
 977         asyncRequest(r, null);
 978     }
 979
 980     @Override
 981     public void asyncRequest(WriteOnly r) {
 982         asyncRequest(r, null);
 983     }
 984
 985     @Override
 986     public <T> T getService(Class<T> api) {
 987         return getSession().getService(api);
 988     }
 989
 990     @Override
 991     public <T> T peekService(Class<T> api) {
 992         return getSession().peekService(api);
 993     }
 994
 995     @Override
 996     public boolean hasService(Class<?> api) {
 997         return getSession().hasService(api);
 998     }
 999
1000     @Override
1001     public <T> void registerService(Class<T> api, T service) {
1002         getSession().registerService(api, service);
1003     }
1004
1005 //    @Override
1006 //    public <T> T syncRequest(Read<T> arg0, AsyncListener<T> arg1) {
1007 //        throw new UnsupportedOperationException("Not implemented.");
1008 //    }
1009
1010     @Override
1011     public <T> T syncRequest(Read<T> arg0, SyncListener<T> arg1) {
1012         throw new UnsupportedOperationException("Not implemented.");
1013     }
1014
1015     @Override
1016     public <T> T syncRequest(Read<T> arg0, Listener<T> arg1) {
1017         throw new UnsupportedOperationException("Not implemented.");
1018     }
1019
1020     @Override
1021     public <T> T syncRequest(Read<T> arg0, SyncProcedure<T> arg1) {
1022         throw new UnsupportedOperationException("Not implemented.");
1023     }
1024
1025     @Override
1026     public <T> T syncRequest(AsyncRead<T> arg0, AsyncListener<T> arg1) {
1027         throw new UnsupportedOperationException("Not implemented.");
1028     }
1029
1030     @Override
1031     public <T> T syncRequest(AsyncRead<T> arg0, SyncListener<T> arg1) {
1032         throw new UnsupportedOperationException("Not implemented.");
1033
1034     }
1035
1036     @Override
1037     public <T> T syncRequest(AsyncRead<T> arg0, Listener<T> arg1) {
1038         throw new UnsupportedOperationException("Not implemented.");
1039
1040     }
1041
1042     @Override
1043     public <T> T syncRequest(AsyncRead<T> arg0, SyncProcedure<T> arg1) {
1044         throw new UnsupportedOperationException("Not implemented.");
1045
1046     }
1047
1048     @Override
1049     public <T> void asyncRequest(Read<T> arg0, AsyncListener<T> arg1) {
1050         throw new UnsupportedOperationException("Not implemented.");
1051
1052     }
1053
1054     @Override
1055     public <T> void asyncRequest(Read<T> arg0, SyncListener<T> arg1) {
1056         throw new UnsupportedOperationException("Not implemented.");
1057
1058     }
1059
1060     @Override
1061     public <T> void asyncRequest(Read<T> arg0, Listener<T> arg1) {
1062         throw new UnsupportedOperationException("Not implemented.");
1063
1064     }
1065
1066     @Override
1067     public <T> void asyncRequest(Read<T> arg0, SyncProcedure<T> arg1) {
1068         throw new UnsupportedOperationException("Not implemented.");
1069
1070     }
1071
1072     @Override
1073     public <T> void asyncRequest(AsyncRead<T> arg0, AsyncListener<T> arg1) {
1074         throw new UnsupportedOperationException("Not implemented.");
1075
1076     }
1077
1078     @Override
1079     public <T> void asyncRequest(AsyncRead<T> arg0, SyncListener<T> arg1) {
1080         throw new UnsupportedOperationException("Not implemented.");
1081
1082     }
1083
1084     @Override
1085     public <T> void asyncRequest(AsyncRead<T> arg0, Listener<T> arg1) {
1086         throw new UnsupportedOperationException("Not implemented.");
1087
1088     }
1089
1090     @Override
1091     public <T> void asyncRequest(AsyncRead<T> arg0, SyncProcedure<T> arg1) {
1092         throw new UnsupportedOperationException("Not implemented.");
1093
1094     }
1095
1096     @Override
1097     public <T> Collection<T> syncRequest(MultiRead<T> arg0, SyncMultiListener<T> arg1) {
1098         throw new UnsupportedOperationException("Not implemented.");
1099
1100     }
1101
1102     @Override
1103     public <T> Collection<T> syncRequest(MultiRead<T> arg0, MultiListener<T> arg1) {
1104         throw new UnsupportedOperationException("Not implemented.");
1105
1106     }
1107
1108     @Override
1109     public <T> Collection<T> syncRequest(MultiRead<T> arg0, SyncMultiProcedure<T> arg1) {
1110         throw new UnsupportedOperationException("Not implemented.");
1111
1112     }
1113
1114     @Override
1115     public <T> Collection<T> syncRequest(AsyncMultiRead<T> arg0,
1116             AsyncMultiListener<T> arg1) {
1117         throw new UnsupportedOperationException("Not implemented.");
1118
1119     }
1120
1121     @Override
1122     public <T> Collection<T> syncRequest(AsyncMultiRead<T> arg0, SyncMultiListener<T> arg1) {
1123         throw new UnsupportedOperationException("Not implemented.");
1124
1125     }
1126
1127     @Override
1128     public <T> Collection<T> syncRequest(AsyncMultiRead<T> arg0,
1129             MultiListener<T> arg1) {
1130         throw new UnsupportedOperationException("Not implemented.");
1131
1132     }
1133
1134     @Override
1135     public <T> Collection<T> syncRequest(AsyncMultiRead<T> arg0, SyncMultiProcedure<T> arg1) {
1136         throw new UnsupportedOperationException("Not implemented.");
1137
1138     }
1139
1140     @Override
1141     public <T> void asyncRequest(MultiRead<T> arg0, SyncMultiListener<T> arg1) {
1142         throw new UnsupportedOperationException("Not implemented.");
1143
1144     }
1145
1146     @Override
1147     public <T> void asyncRequest(MultiRead<T> arg0, MultiListener<T> arg1) {
1148         throw new UnsupportedOperationException("Not implemented.");
1149
1150     }
1151
1152     @Override
1153     public <T> void asyncRequest(MultiRead<T> arg0, SyncMultiProcedure<T> arg1) {
1154         throw new UnsupportedOperationException("Not implemented.");
1155
1156     }
1157
1158     @Override
1159     public <T> void asyncRequest(AsyncMultiRead<T> arg0,
1160             AsyncMultiListener<T> arg1) {
1161         throw new UnsupportedOperationException("Not implemented.");
1162     }
1163
1164     @Override
1165     public <T> void asyncRequest(AsyncMultiRead<T> arg0, SyncMultiListener<T> arg1) {
1166         throw new UnsupportedOperationException("Not implemented.");
1167
1168     }
1169
1170     @Override
1171     public <T> void asyncRequest(AsyncMultiRead<T> arg0,
1172             MultiListener<T> arg1) {
1173         throw new UnsupportedOperationException("Not implemented.");
1174
1175     }
1176
1177     @Override
1178     public <T> void asyncRequest(AsyncMultiRead<T> arg0, SyncMultiProcedure<T> arg1) {
1179         throw new UnsupportedOperationException("Not implemented.");
1180     }
1181
1182     @Override
1183     public <T> void asyncRequest(ExternalRead<T> request, Procedure<T> procedure) {
1184         throw new UnsupportedOperationException("Not implemented.");
1185     }
1186
1187     @Override
1188     public <T> T syncRequest(ExternalRead<T> request) {
1189         throw new UnsupportedOperationException("Not implemented.");
1190     }
1191
1192     @Override
1193     public <T> T syncRequest(ExternalRead<T> request, Listener<T> procedure) {
1194         throw new UnsupportedOperationException("Not implemented.");
1195     }
1196
1197     @Override
1198     public <T> T syncRequest(ExternalRead<T> request, Procedure<T> procedure) {
1199         throw new UnsupportedOperationException("Not implemented.");
1200     }
1201
1202     @Override
1203     public <T> void asyncRequest(ExternalRead<T> request) {
1204         throw new UnsupportedOperationException("Not implemented.");
1205     }
1206
1207     @Override
1208     public <T> void asyncRequest(ExternalRead<T> request, Listener<T> procedure) {
1209         throw new UnsupportedOperationException("Not implemented.");
1210     }
1211
1212     @Override
1213     public void syncRequest(DelayedWrite request) throws DatabaseException {
1214         throw new UnsupportedOperationException("Not implemented.");
1215     }
1216
1217         @Override
1218         public <T> T syncRequest(WriteResult<T> request) throws DatabaseException {
1219                 throw new UnsupportedOperationException();
1220         }
1221
1222         @Override
1223         public <T> T syncRequest(DelayedWriteResult<T> request)
1224                         throws DatabaseException {
1225                 throw new UnsupportedOperationException();
1226         }
1227
1228         @Override
1229         public <T> T syncRequest(WriteOnlyResult<T> r) throws DatabaseException {
1230                 throw new UnsupportedOperationException();
1231         }
1232
1233         @Override
1234         public <T> void asyncRequest(WriteResult<T> r, Procedure<T> procedure) {
1235                 throw new UnsupportedOperationException();
1236         }
1237
1238         @Override
1239         public <T> void asyncRequest(DelayedWriteResult<T> r, Procedure<T> procedure) {
1240                 throw new UnsupportedOperationException();
1241         }
1242
1243         @Override
1244         public <T> void asyncRequest(WriteOnlyResult<T> r, Procedure<T> procedure) {
1245                 throw new UnsupportedOperationException();
1246         }
1247
1248         @Override
1249         public Resource getRootLibrary() {
1250                 return processor.getRootLibrary();
1251         }
1252
1253         @Override
1254         public <T> void async(ReadInterface<T> r, Procedure<T> procedure) {
1255                 throw new UnsupportedOperationException();
1256         }
1257
1258         @Override
1259         public <T> void async(ReadInterface<T> r, AsyncProcedure<T> procedure) {
1260                 throw new UnsupportedOperationException();
1261         }
1262
1263         @Override
1264         public <T> void async(ReadInterface<T> r, SyncProcedure<T> procedure) {
1265                 throw new UnsupportedOperationException();
1266         }
1267
1268         @Override
1269         public <T> void async(ReadInterface<T> r, Listener<T> procedure) {
1270                 throw new UnsupportedOperationException();
1271         }
1272
1273         @Override
1274         public <T> void async(ReadInterface<T> r, AsyncListener<T> procedure) {
1275                 throw new UnsupportedOperationException();
1276         }
1277
1278         @Override
1279         public <T> void async(ReadInterface<T> r, SyncListener<T> procedure) {
1280                 throw new UnsupportedOperationException();
1281         }
1282
1283         @Override
1284         public <T> T sync(ReadInterface<T> r) throws DatabaseException {
1285                 throw new UnsupportedOperationException();
1286         }
1287
1288         @Override
1289         public <T> T sync(WriteInterface<T> r) throws DatabaseException {
1290                 throw new UnsupportedOperationException();
1291         }
1292
1293         @Override
1294         public <T> void async(WriteInterface<T> r, Procedure<T> procedure) {
1295                 throw new UnsupportedOperationException();
1296         }
1297
1298         @Override
1299         public <T> void async(WriteInterface<T> r) {
1300                 throw new UnsupportedOperationException();
1301         }
1302
1303         @Override
1304         public Object getModificationCounter() {
1305                 throw new UnsupportedOperationException();
1306         }
1307
1308 }