--- /dev/null
+/*******************************************************************************\r
+ * Copyright (c) 2007, 2010 Association for Decentralized Information Management\r
+ * in Industry THTH ry.\r
+ * All rights reserved. This program and the accompanying materials\r
+ * are made available under the terms of the Eclipse Public License v1.0\r
+ * which accompanies this distribution, and is available at\r
+ * http://www.eclipse.org/legal/epl-v10.html\r
+ *\r
+ * Contributors:\r
+ * VTT Technical Research Centre of Finland - initial API and implementation\r
+ *******************************************************************************/\r
+package org.simantics.db.common.processor;\r
+\r
+import java.util.ArrayList;\r
+import java.util.Collection;\r
+import java.util.HashMap;\r
+import java.util.HashSet;\r
+import java.util.LinkedList;\r
+import java.util.Set;\r
+import java.util.UUID;\r
+import java.util.concurrent.Semaphore;\r
+\r
+import org.simantics.db.AsyncReadGraph;\r
+import org.simantics.db.AsyncRequestProcessor;\r
+import org.simantics.db.ReadGraph;\r
+import org.simantics.db.RequestProcessor;\r
+import org.simantics.db.Resource;\r
+import org.simantics.db.Session;\r
+import org.simantics.db.VirtualGraph;\r
+import org.simantics.db.WriteGraph;\r
+import org.simantics.db.common.procedure.adapter.AsyncMultiProcedureAdapter;\r
+import org.simantics.db.common.procedure.adapter.ProcedureAdapter;\r
+import org.simantics.db.common.procedure.wrapper.NoneToAsyncProcedure;\r
+import org.simantics.db.common.request.ReadRequest;\r
+import org.simantics.db.common.request.WriteRequest;\r
+import org.simantics.db.common.utils.Logger;\r
+import org.simantics.db.exception.CancelTransactionException;\r
+import org.simantics.db.exception.DatabaseException;\r
+import org.simantics.db.procedure.AsyncListener;\r
+import org.simantics.db.procedure.AsyncMultiListener;\r
+import org.simantics.db.procedure.AsyncMultiProcedure;\r
+import org.simantics.db.procedure.AsyncProcedure;\r
+import org.simantics.db.procedure.Listener;\r
+import org.simantics.db.procedure.MultiListener;\r
+import org.simantics.db.procedure.MultiProcedure;\r
+import org.simantics.db.procedure.Procedure;\r
+import org.simantics.db.procedure.SyncListener;\r
+import org.simantics.db.procedure.SyncMultiListener;\r
+import org.simantics.db.procedure.SyncMultiProcedure;\r
+import org.simantics.db.procedure.SyncProcedure;\r
+import org.simantics.db.request.AsyncMultiRead;\r
+import org.simantics.db.request.AsyncRead;\r
+import org.simantics.db.request.DelayedWrite;\r
+import org.simantics.db.request.DelayedWriteResult;\r
+import org.simantics.db.request.ExternalRead;\r
+import org.simantics.db.request.MultiRead;\r
+import org.simantics.db.request.Read;\r
+import org.simantics.db.request.ReadInterface;\r
+import org.simantics.db.request.UndoTraits;\r
+import org.simantics.db.request.Write;\r
+import org.simantics.db.request.WriteInterface;\r
+import org.simantics.db.request.WriteOnly;\r
+import org.simantics.db.request.WriteOnlyResult;\r
+import org.simantics.db.request.WriteResult;\r
+import org.simantics.utils.DataContainer;\r
+import org.simantics.utils.datastructures.Callback;\r
+import org.simantics.utils.datastructures.Pair;\r
+\r
+public class MergingGraphRequestProcessor implements RequestProcessor {\r
+\r
+ private static class SyncWriteRequestAdapter implements Write {\r
+\r
+ private Semaphore semaphore = new Semaphore(0);\r
+ private Object request;\r
+ private Throwable exception;\r
+ SyncWriteRequestAdapter(Write r) {\r
+ this.request = r;\r
+ }\r
+ SyncWriteRequestAdapter(WriteOnly r) {\r
+ this.request = r;\r
+ }\r
+// @Override\r
+// public GraphRequestStatus perform(Graph g) throws Exception {\r
+// return perform((ReadGraph)g);\r
+// }\r
+ @Override\r
+ public void perform(WriteGraph g) throws DatabaseException, CancelTransactionException {\r
+ if(request instanceof Write) {\r
+ ((Write)request).perform(g);\r
+ } else if(request instanceof DelayedWrite) {\r
+ ((DelayedWrite)request).perform(g);\r
+ } else {\r
+ ((WriteOnly)request).perform(g);\r
+ }\r
+ }\r
+// @Override\r
+// public String getId() {\r
+// if(request instanceof WriteGraphRequest) {\r
+// return ((WriteGraphRequest)request).getId();\r
+// } else {\r
+// return null;\r
+// }\r
+// }\r
+// @Override\r
+// public void requestCompleted(GraphRequestStatus status) {\r
+// if(request instanceof WriteGraphRequest) {\r
+// ((WriteGraphRequest)request).requestCompleted(status);\r
+// } else {\r
+// }\r
+// }\r
+// @Override\r
+// public void handleException(Throwable e) {\r
+// this.exception = e;\r
+// if(request instanceof WriteGraphRequest) {\r
+// ((WriteGraphRequest)request).handleException(e);\r
+// }\r
+// }\r
+\r
+ public void throwOrWrapException() {\r
+ if (exception == null)\r
+ return;\r
+ if (exception instanceof RuntimeException)\r
+ throw (RuntimeException) exception;\r
+ if (exception instanceof Error)\r
+ throw (Error) exception;\r
+ throw new RuntimeException("See cause for the real exception.", exception);\r
+ }\r
+\r
+ @Override\r
+ public VirtualGraph getProvider() {\r
+ return null;\r
+ }\r
+\r
+// @Override\r
+// public void fillMetadata(Map<String, String> metadata) {\r
+// }\r
+\r
+ public void acquire() {\r
+ try {\r
+ semaphore.acquire();\r
+ } catch (InterruptedException e) {\r
+ Logger.defaultLogError(e);\r
+ }\r
+ }\r
+\r
+ public void release() {\r
+ semaphore.release();\r
+ }\r
+\r
+ @Override\r
+ public UndoTraits getUndoTraits() {\r
+ return null;\r
+ }\r
+ \r
+ @Override\r
+ public String toString() {\r
+ return "SyncWriteRequestAdapter " + request;\r
+ }\r
+\r
+ }\r
+\r
+ long transactionKeepalivePeriod;\r
+\r
+ /**\r
+ * Synchronization object for implementing {@link #synchronize()}.\r
+ * {@link Object#notifyAll()} is invoked for this lock object every time a\r
+ * single transaction is completed, thereby releasing all waiters in\r
+ * {@link #synchronize()}.\r
+ */\r
+ Object barrier = new Object();\r
+\r
+ Set<Pair<Object, Object>> requestSet = new HashSet<Pair<Object, Object>>();\r
+ LinkedList<Pair<Object, Object>> requestQueue = new LinkedList<Pair<Object, Object>>();\r
+ boolean hasAlreadyRequest = false;\r
+\r
+ /**\r
+ * A set of requests which {@link #synchronize()} is depending on at the\r
+ * moment. Every time a request within this set is completed, some thread in\r
+ * {@link #synchronize()} should be released.\r
+ */\r
+// Set<Object> barrierRequests = new HashSet<Object>();\r
+ Set<Object> syncRequests = new HashSet<Object>();\r
+\r
+ private String name;\r
+\r
+ private AsyncRequestProcessor processor;\r
+\r
+ public MergingGraphRequestProcessor(String name, AsyncRequestProcessor processor, long transactionKeepalivePeriod) {\r
+ this.name = name;\r
+ this.processor = processor;\r
+ this.transactionKeepalivePeriod = transactionKeepalivePeriod;\r
+ }\r
+\r
+ public MergingGraphRequestProcessor(AsyncRequestProcessor processor, long transactionKeepalivePeriod) {\r
+ this.name = "MergingGraphRequestProcessor" + UUID.randomUUID().toString();\r
+ this.processor = processor;\r
+ this.transactionKeepalivePeriod = transactionKeepalivePeriod;\r
+ }\r
+\r
+ @SuppressWarnings({"unchecked", "rawtypes"})\r
+ protected class MergedRead extends ReadRequest {\r
+\r
+ Pair<Object, Object> currentRequest;\r
+\r
+// RunnerReadGraphRequest(GraphRequestProcessor processor) {\r
+// super(processor);\r
+// }\r
+//\r
+// @Override\r
+// public void completed(boolean value) {\r
+//// System.out.println(this + "MGRP read completed");\r
+//// synchronized (MergingGraphRequestProcessor.this) {\r
+//// if (requestQueue.isEmpty())\r
+//// hasAlreadyRequest = false;\r
+//// else\r
+//// newTransaction();\r
+//// }\r
+// }\r
+\r
+ @Override\r
+ public void run(ReadGraph graph) {\r
+\r
+// System.out.println(MergingGraphRequestProcessor.this + " reads");\r
+\r
+ while (true) {\r
+\r
+ synchronized (MergingGraphRequestProcessor.this) {\r
+\r
+ // Release #synchronize() invokers if necessary.\r
+// if (currentRequest != null && barrierRequests.contains(currentRequest)) {\r
+// synchronized (barrier) {\r
+// barrier.notifyAll();\r
+// }\r
+// }\r
+\r
+ if(requestQueue.isEmpty()) {\r
+ if (transactionKeepalivePeriod > 0) {\r
+// System.out.println("MGRP [" + MergingGraphRequestProcessor.this + "] waits " + transactionKeepalivePeriod + " ms. in " + Thread.currentThread() );\r
+ try {\r
+ MergingGraphRequestProcessor.this.wait(transactionKeepalivePeriod);\r
+ } catch (InterruptedException e) {\r
+ Logger.defaultLogError(e);\r
+ }\r
+ if (requestQueue.isEmpty())\r
+ break;\r
+ } else\r
+ break;\r
+ }\r
+\r
+ Object nextRequest = requestQueue.peekFirst().first;\r
+ if(nextRequest instanceof Write || nextRequest instanceof DelayedWrite) {\r
+ break;\r
+ }\r
+\r
+ currentRequest = requestQueue.remove(0);\r
+ requestSet.remove(currentRequest);\r
+\r
+ }\r
+\r
+// ReadGraphRequest req = (ReadGraphRequest)currentRequest.first;\r
+\r
+ if( syncRequests.contains(currentRequest.first)) {\r
+\r
+ try {\r
+\r
+ if(currentRequest.second instanceof AsyncProcedure<?>) {\r
+ if(currentRequest.first instanceof Read) {\r
+ Read req = (Read)currentRequest.first;\r
+ graph.syncRequest(req, (AsyncProcedure<?>)currentRequest.second);\r
+ } else {\r
+ AsyncRead req = (AsyncRead)currentRequest.first;\r
+ graph.syncRequest(req, (AsyncProcedure<?>)currentRequest.second);\r
+ }\r
+ } else {\r
+ AsyncMultiRead req = (AsyncMultiRead)currentRequest.first;\r
+ graph.syncRequest(req, (AsyncMultiProcedure<?>)currentRequest.second);\r
+ }\r
+\r
+ } catch(Throwable t) {\r
+\r
+ Logger.defaultLogError(t);\r
+\r
+ if(currentRequest.second instanceof AsyncProcedure<?>) {\r
+ ((AsyncProcedure<?>)currentRequest.second).exception(graph, t);\r
+ } else {\r
+ ((AsyncMultiProcedure<?>)currentRequest.second).exception(graph, t);\r
+ }\r
+\r
+ }\r
+\r
+ synchronized (currentRequest.first) {\r
+ syncRequests.remove(currentRequest.first);\r
+// System.out.println("notifying " + currentRequest.first);\r
+ currentRequest.first.notify();\r
+ }\r
+\r
+\r
+ } else {\r
+\r
+ try{\r
+\r
+ if(currentRequest.second instanceof AsyncProcedure<?>) {\r
+ if(currentRequest.first instanceof AsyncRead) {\r
+ AsyncRead req = (AsyncRead)currentRequest.first;\r
+ graph.asyncRequest(req, (AsyncProcedure<?>)currentRequest.second);\r
+ } else {\r
+ Read req = (Read)currentRequest.first;\r
+ graph.asyncRequest(req, (AsyncProcedure<?>)currentRequest.second);\r
+ }\r
+ } else {\r
+ AsyncMultiRead req = (AsyncMultiRead)currentRequest.first;\r
+ graph.asyncRequest(req, (AsyncMultiProcedure<?>)currentRequest.second);\r
+ }\r
+\r
+ } catch(Throwable t) {\r
+\r
+ Logger.defaultLogError(t);\r
+\r
+ if(currentRequest.second instanceof AsyncProcedure<?>) {\r
+ ((AsyncProcedure<?>)currentRequest.second).exception(graph, t);\r
+ } else {\r
+ ((AsyncMultiProcedure<?>)currentRequest.second).exception(graph, t);\r
+ }\r
+\r
+ }\r
+ }\r
+\r
+ }\r
+\r
+// System.out.println(MergingGraphRequestProcessor.this + " read completed");\r
+\r
+ synchronized (MergingGraphRequestProcessor.this) {\r
+ if (requestQueue.isEmpty())\r
+ hasAlreadyRequest = false;\r
+ else\r
+ newTransaction();\r
+ }\r
+\r
+ }\r
+\r
+ @Override\r
+ public String toString() {\r
+ return "MergedRead[" + requestQueue.size() + " requests]";\r
+ }\r
+\r
+ }\r
+\r
+ protected class RunnerWriteGraphRequest extends WriteRequest {\r
+\r
+ Pair<Object, Object> currentRequest;\r
+ HashMap<String, String> metadata = new HashMap<String, String>();\r
+\r
+ @Override\r
+ public void perform(WriteGraph graph) throws DatabaseException {\r
+\r
+// System.out.println(MergingGraphRequestProcessor.this + " writes");\r
+\r
+ while (true) {\r
+\r
+ synchronized (MergingGraphRequestProcessor.this) {\r
+\r
+ // Release #synchronize() invokers if necessary.\r
+// if (currentRequest != null && barrierRequests.contains(currentRequest)) {\r
+// synchronized (barrier) {\r
+// barrier.notifyAll();\r
+// }\r
+// }\r
+\r
+ if(requestQueue.isEmpty()) {\r
+ if (transactionKeepalivePeriod > 0) {\r
+ try {\r
+ MergingGraphRequestProcessor.this.wait(transactionKeepalivePeriod);\r
+ } catch (InterruptedException e) {\r
+ Logger.defaultLogError(e);\r
+ }\r
+ if (requestQueue.isEmpty())\r
+ break;\r
+ } else\r
+ break;\r
+ }\r
+\r
+ Object nextRequest = requestQueue.peekFirst().first;\r
+ if(nextRequest instanceof AsyncMultiRead || nextRequest instanceof AsyncRead || nextRequest instanceof Read) {\r
+ break;\r
+ }\r
+\r
+ currentRequest = requestQueue.remove(0);\r
+ requestSet.remove(currentRequest);\r
+\r
+ }\r
+\r
+ @SuppressWarnings("unchecked")\r
+ Callback<Throwable> callback = (Callback<Throwable>)currentRequest.second;\r
+\r
+ if (currentRequest.first.getClass().equals(SyncWriteRequestAdapter.class)) {\r
+\r
+ SyncWriteRequestAdapter adapter = (SyncWriteRequestAdapter)currentRequest.first;\r
+\r
+ try {\r
+// System.out.println("merg.sync " + adapter);\r
+ graph.syncRequest(adapter);\r
+ if(callback != null) callback.run(null);\r
+ } catch(Throwable t) {\r
+ Logger.defaultLogError(t);\r
+ if(callback != null) callback.run(t);\r
+ }\r
+\r
+ adapter.release();\r
+// System.out.println("merg.sync.release " + adapter);\r
+\r
+ } else {\r
+\r
+ try {\r
+ if(currentRequest.first instanceof Write) graph.syncRequest((Write)currentRequest.first); \r
+ else if(currentRequest.first instanceof DelayedWrite) graph.syncRequest((DelayedWrite)currentRequest.first); \r
+ if(callback != null) callback.run(null);\r
+ } catch(Throwable t) {\r
+ Logger.defaultLogError(t);\r
+ if(callback != null) callback.run(t);\r
+ }\r
+\r
+ }\r
+\r
+ }\r
+\r
+// System.out.println(MergingGraphRequestProcessor.this + " write completed");\r
+\r
+ synchronized (MergingGraphRequestProcessor.this) {\r
+ if (requestQueue.isEmpty())\r
+ hasAlreadyRequest = false;\r
+ else\r
+ newTransaction();\r
+ }\r
+\r
+ }\r
+\r
+ }\r
+\r
+ private void newTransaction() {\r
+\r
+ boolean write = false;\r
+\r
+ synchronized (MergingGraphRequestProcessor.this) {\r
+ assert(!requestQueue.isEmpty());\r
+ Object nextRequest = requestQueue.peekFirst().first;\r
+ write = (nextRequest instanceof Write || nextRequest instanceof DelayedWrite);\r
+ }\r
+\r
+ if(write) {\r
+ processor.asyncRequest(new RunnerWriteGraphRequest(), null);\r
+ } else {\r
+ processor.asyncRequest(new MergedRead());\r
+ }\r
+\r
+ }\r
+\r
+ @Override\r
+ public <T> void asyncRequest(AsyncMultiRead<T> request, AsyncMultiProcedure<T> procedure) {\r
+\r
+// System.out.println(this + " asyncRequest(ReadGraphRequest<QueryProcedure4<T>> request, QueryProcedure4<T> procedure)");\r
+\r
+ if (requestSet.contains(request))\r
+ return;\r
+\r
+ Pair<Object, Object> pair = new Pair<Object, Object>(request, procedure);\r
+ requestQueue.add(pair);\r
+ requestSet.add(pair);\r
+\r
+ if (!hasAlreadyRequest) {\r
+ newTransaction();\r
+ hasAlreadyRequest = true;\r
+ } else {\r
+ notify();\r
+ }\r
+\r
+ }\r
+\r
+ @Override\r
+ public synchronized <T> void asyncRequest(AsyncRead<T> request, AsyncProcedure<T> procedure) {\r
+\r
+// System.out.println(this + " asyncRequest(ReadGraphRequest<SingleQueryProcedure4<T>> request, SingleQueryProcedure4<T> procedure) " + this);\r
+\r
+ if (requestSet.contains(request))\r
+ return;\r
+\r
+ Pair<Object, Object> pair = new Pair<Object, Object>(request, procedure);\r
+ requestQueue.add(pair);\r
+ requestSet.add(pair);\r
+\r
+ if (!hasAlreadyRequest) {\r
+ newTransaction();\r
+ hasAlreadyRequest = true;\r
+ } else {\r
+// System.out.println("notify " + this);\r
+ notify();\r
+ }\r
+\r
+ }\r
+\r
+ @Override\r
+ public synchronized void asyncRequest(Write request, Callback<DatabaseException> callback) {\r
+\r
+// System.out.println(this + " asyncRequest(WriteGraphRequest request)");\r
+\r
+ if (requestSet.contains(request))\r
+ return;\r
+\r
+ Pair<Object, Object> pair = new Pair<Object, Object>(request, callback);\r
+ requestQueue.add(pair);\r
+ requestSet.add(pair);\r
+\r
+ if (!hasAlreadyRequest) {\r
+// System.out.println("new transaction");\r
+ newTransaction();\r
+ hasAlreadyRequest = true;\r
+ } else {\r
+// System.out.println("notify");\r
+ notify();\r
+ }\r
+\r
+ }\r
+\r
+ @Override\r
+ public synchronized void asyncRequest(DelayedWrite request, Callback<DatabaseException> callback) {\r
+\r
+// System.out.println(this + " asyncRequest(WriteGraphRequest request)");\r
+\r
+ if (requestSet.contains(request))\r
+ return;\r
+\r
+ Pair<Object, Object> pair = new Pair<Object, Object>(request, callback);\r
+ requestQueue.add(pair);\r
+ requestSet.add(pair);\r
+\r
+ if (!hasAlreadyRequest) {\r
+// System.out.println("new transaction");\r
+ newTransaction();\r
+ hasAlreadyRequest = true;\r
+ } else {\r
+// System.out.println("notify");\r
+ notify();\r
+ }\r
+\r
+ }\r
+\r
+ @Override\r
+ public synchronized void asyncRequest(WriteOnly request, Callback<DatabaseException> callback) {\r
+\r
+// System.out.println(this + " asyncRequest(WriteGraphRequest request)");\r
+\r
+ if (requestSet.contains(request))\r
+ return;\r
+\r
+ Pair<Object, Object> pair = new Pair<Object, Object>(request, callback);\r
+ requestQueue.add(pair);\r
+ requestSet.add(pair);\r
+\r
+ if (!hasAlreadyRequest) {\r
+// System.out.println("new transaction");\r
+ newTransaction();\r
+ hasAlreadyRequest = true;\r
+ } else {\r
+// System.out.println("notify");\r
+ notify();\r
+ }\r
+\r
+ }\r
+\r
+ @Override\r
+ public <T> Collection<T> syncRequest(AsyncMultiRead<T> request, final AsyncMultiProcedure<T> procedure) {\r
+\r
+ final DataContainer<Throwable> throwable = new DataContainer<Throwable>(null);\r
+\r
+ // Queue the adapter up for execution.\r
+ synchronized (request) {\r
+ syncRequests.add(request);\r
+ asyncRequest(request, procedure);\r
+ if(syncRequests.contains(request)) {\r
+ try {\r
+ // System.out.println("waiting " + request);\r
+ request.wait();\r
+ } catch (InterruptedException e) {\r
+ throw new Error(e);\r
+ }\r
+ }\r
+ }\r
+\r
+ Throwable t = throwable.get();\r
+\r
+ if(t != null) {\r
+ Logger.defaultLogError(t);\r
+ throw new RuntimeException(t.getMessage());\r
+ }\r
+ \r
+ return null;\r
+\r
+ }\r
+\r
+ @Override\r
+ public <T> T syncRequest(AsyncRead<T> request, final AsyncProcedure<T> procedure) {\r
+\r
+// System.out.println("syncRequest(ReadGraphRequest<SingleQueryProcedure4<T>> request, SingleQueryProcedure4<T> procedure)");\r
+\r
+ final DataContainer<T> result = new DataContainer<T>(null);\r
+ final DataContainer<Throwable> throwable = new DataContainer<Throwable>(null);\r
+\r
+ // Queue the adapter up for execution.\r
+ synchronized (request) {\r
+\r
+ syncRequests.add(request);\r
+ asyncRequest(request, new AsyncProcedure<T>() {\r
+\r
+ public void execute(AsyncReadGraph graph, T t) {\r
+ synchronized(result) {\r
+ result.set(t);\r
+ }\r
+ procedure.execute(graph, t);\r
+ };\r
+\r
+ @Override\r
+ public void exception(AsyncReadGraph graph, Throwable t) {\r
+ throwable.set(t);\r
+ }\r
+\r
+ @Override\r
+ public String toString() {\r
+ return procedure.toString();\r
+ }\r
+\r
+ });\r
+ if(syncRequests.contains(request)) {\r
+ try {\r
+ // System.out.println("waiting " + request);\r
+ request.wait();\r
+ } catch (InterruptedException e) {\r
+ throw new Error(e);\r
+ }\r
+ }\r
+ }\r
+\r
+ Throwable t = throwable.get();\r
+\r
+ if(t != null) {\r
+ Logger.defaultLogError(t);\r
+ throw new RuntimeException(t.getMessage());\r
+ }\r
+ \r
+ return result.get();\r
+\r
+ //return result.get();\r
+\r
+ }\r
+\r
+\r
+ @Override\r
+ public void syncRequest(Write request) {\r
+\r
+// System.out.println(MergingGraphRequestProcessor.this + " syncRequest(WriteGraphRequest)");\r
+\r
+ SyncWriteRequestAdapter adapter = new SyncWriteRequestAdapter(request);\r
+\r
+ asyncRequest(adapter, null);\r
+\r
+ adapter.acquire();\r
+\r
+ // Throw exception if one occurred.\r
+ adapter.throwOrWrapException();\r
+\r
+ }\r
+\r
+ @Override\r
+ public void syncRequest(WriteOnly request) {\r
+\r
+// System.out.println(MergingGraphRequestProcessor.this + " syncRequest(WriteGraphRequest)");\r
+\r
+ SyncWriteRequestAdapter adapter = new SyncWriteRequestAdapter(request);\r
+\r
+ // Queue the adapter up for execution.\r
+ synchronized (adapter) {\r
+ asyncRequest(adapter, null);\r
+ try {\r
+ adapter.wait();\r
+ } catch (InterruptedException e) {\r
+ throw new Error(e);\r
+ }\r
+ }\r
+\r
+ // Throw exception if one occurred.\r
+ adapter.throwOrWrapException();\r
+\r
+ }\r
+\r
+ @Override\r
+ public Session getSession() {\r
+ return processor.getSession();\r
+ }\r
+\r
+ @Override\r
+ public String toString() {\r
+ return "MergingGraphRequestProcessor[" + name + "]@" + System.identityHashCode(this) + " (based on " + processor + ")";\r
+ }\r
+\r
+ @Override\r
+ public <T> void asyncRequest(AsyncRead<T> request) {\r
+\r
+ asyncRequest(request, new ProcedureAdapter<T>() {\r
+\r
+ @Override\r
+ public void exception(Throwable t) {\r
+ Logger.defaultLogError(t);\r
+ }\r
+\r
+ });\r
+\r
+ }\r
+\r
+ @Override\r
+ public <T> void asyncRequest(AsyncRead<T> request, Procedure<T> procedure) {\r
+ asyncRequest(request, new NoneToAsyncProcedure<T>(procedure));\r
+ }\r
+\r
+ @Override\r
+ public <T> void asyncRequest(AsyncMultiRead<T> request) {\r
+ throw new UnsupportedOperationException("Not implemented.");\r
+ }\r
+\r
+ @Override\r
+ public <T> void asyncRequest(AsyncMultiRead<T> request,\r
+ MultiProcedure<T> procedure) {\r
+ throw new UnsupportedOperationException("Not implemented.");\r
+ }\r
+\r
+ @Override\r
+ public <T> T syncRequest(AsyncRead<T> request) {\r
+\r
+ final DataContainer<Throwable> throwable = new DataContainer<Throwable>(null);\r
+ final DataContainer<T> result = new DataContainer<T>();\r
+\r
+ syncRequest(request, new AsyncProcedure<T>() {\r
+\r
+ public void execute(AsyncReadGraph graph, T t) {\r
+ result.set(t);\r
+ }\r
+\r
+ @Override\r
+ public void exception(AsyncReadGraph graph, Throwable t) {\r
+ throwable.set(t);\r
+ }\r
+\r
+ });\r
+\r
+ Throwable t = throwable.get();\r
+\r
+ if(t != null) {\r
+ Logger.defaultLogError(t);\r
+ throw new RuntimeException(t.getMessage());\r
+ }\r
+\r
+ return result.get();\r
+\r
+ }\r
+\r
+ @Override\r
+ public <T> T syncRequest(AsyncRead<T> request,\r
+ Procedure<T> procedure) {\r
+ throw new UnsupportedOperationException("Not implemented.");\r
+ }\r
+\r
+ @Override\r
+ public <T> Collection<T> syncRequest(AsyncMultiRead<T> request) {\r
+\r
+ final DataContainer<Throwable> throwable = new DataContainer<Throwable>(null);\r
+ final ArrayList<T> result = new ArrayList<T>();\r
+\r
+ syncRequest(request, new AsyncMultiProcedureAdapter<T>() {\r
+\r
+ @Override\r
+ public void execute(AsyncReadGraph graph, T t) {\r
+ synchronized(result) {\r
+ result.add(t);\r
+ }\r
+ };\r
+\r
+ @Override\r
+ public void exception(AsyncReadGraph graph, Throwable t) {\r
+ throwable.set(t);\r
+ }\r
+\r
+ });\r
+\r
+ Throwable t = throwable.get();\r
+\r
+ if(t != null) {\r
+ Logger.defaultLogError(t);\r
+ throw new RuntimeException(t.getMessage());\r
+ }\r
+\r
+ return result;\r
+\r
+ }\r
+\r
+ @Override\r
+ public <T> Collection<T> syncRequest(AsyncMultiRead<T> request,\r
+ MultiProcedure<T> procedure) {\r
+ throw new Error("Not implemented.");\r
+ }\r
+\r
+ @Override\r
+ public <T> T syncRequest(Read<T> request) {\r
+\r
+ final DataContainer<Throwable> throwable = new DataContainer<Throwable>(null);\r
+ final DataContainer<T> result = new DataContainer<T>();\r
+\r
+\r
+ syncRequest(request, new Procedure<T>() {\r
+\r
+ public void execute(T t) {\r
+ result.set(t);\r
+ }\r
+\r
+ @Override\r
+ public void exception(Throwable t) {\r
+ throwable.set(t);\r
+ }\r
+\r
+ });\r
+\r
+ Throwable t = throwable.get();\r
+\r
+ if(t != null) {\r
+ throw new Error(t.getMessage());\r
+ }\r
+\r
+\r
+ return result.get();\r
+\r
+ }\r
+\r
+ @Override\r
+ public <T> T syncRequest(Read<T> request,\r
+ final AsyncProcedure<T> procedure) {\r
+\r
+ final DataContainer<T> result = new DataContainer<T>(null);\r
+ final DataContainer<Throwable> throwable = new DataContainer<Throwable>(null);\r
+\r
+ // Queue the adapter up for execution.\r
+ synchronized (request) {\r
+\r
+ syncRequests.add(request);\r
+ asyncRequest(request, new AsyncProcedure<T>() {\r
+\r
+ public void execute(AsyncReadGraph graph, T t) {\r
+ synchronized(result) {\r
+ result.set(t);\r
+ }\r
+ procedure.execute(graph, t);\r
+ };\r
+\r
+ @Override\r
+ public void exception(AsyncReadGraph graph, Throwable t) {\r
+ throwable.set(t);\r
+ }\r
+\r
+ @Override\r
+ public String toString() {\r
+ return procedure.toString();\r
+ }\r
+\r
+ });\r
+ if(syncRequests.contains(request)) {\r
+ try {\r
+ // System.out.println("waiting " + request);\r
+ request.wait();\r
+ } catch (InterruptedException e) {\r
+ throw new Error(e);\r
+ }\r
+ }\r
+ }\r
+\r
+ Throwable t = throwable.get();\r
+\r
+ if(t != null) {\r
+ throw new RuntimeException("Unexpected exception in MergingGraphRequestProcessor.syncRequest(Read, AsyncProcedure)", t);\r
+ }\r
+ \r
+ return result.get();\r
+\r
+ }\r
+\r
+ @Override\r
+ public <T> void asyncRequest(Read<T> request) {\r
+\r
+ asyncRequest(request, new ProcedureAdapter<T>() {\r
+\r
+ @Override\r
+ public void exception(Throwable t) {\r
+ Logger.defaultLogError(t);\r
+ }\r
+\r
+ });\r
+\r
+ }\r
+\r
+ @Override\r
+ public synchronized <T> void asyncRequest(Read<T> request,\r
+ AsyncProcedure<T> procedure) {\r
+\r
+ if (requestSet.contains(request))\r
+ return;\r
+\r
+ Pair<Object, Object> pair = new Pair<Object, Object>(request, procedure);\r
+ requestQueue.add(pair);\r
+ requestSet.add(pair);\r
+\r
+ if (!hasAlreadyRequest) {\r
+ newTransaction();\r
+ hasAlreadyRequest = true;\r
+ } else {\r
+ notify();\r
+ }\r
+\r
+ }\r
+\r
+ @Override\r
+ public <T> T syncRequest(Read<T> request, Procedure<T> procedure) {\r
+ return syncRequest(request, new NoneToAsyncProcedure<T>(procedure));\r
+ }\r
+\r
+ @Override\r
+ public <T> Collection<T> syncRequest(final MultiRead<T> request) throws DatabaseException {\r
+ assert(request != null);\r
+\r
+ final ArrayList<T> result = new ArrayList<T>();\r
+ final DataContainer<Throwable> exception = new DataContainer<Throwable>();\r
+\r
+ syncRequest(request, new AsyncMultiProcedureAdapter<T>() {\r
+\r
+ @Override\r
+ public void execute(AsyncReadGraph graph, T t) {\r
+ synchronized(result) {\r
+ result.add(t);\r
+ }\r
+ }\r
+\r
+ @Override\r
+ public void exception(AsyncReadGraph graph, Throwable t) {\r
+ exception.set(t);\r
+ }\r
+\r
+ @Override\r
+ public String toString() {\r
+ return "syncRequest(MultiRead) -> " + request;\r
+ }\r
+\r
+ });\r
+\r
+ Throwable t = exception.get();\r
+ if(t != null) {\r
+ if(t instanceof DatabaseException) throw (DatabaseException)t;\r
+ else throw new DatabaseException("Unexpected exception in ReadGraph.syncRequest(Read)", t);\r
+ }\r
+\r
+ return result;\r
+ }\r
+\r
+ @Override\r
+ public <T> Collection<T> syncRequest(MultiRead<T> request, AsyncMultiProcedure<T> procedure) {\r
+ throw new UnsupportedOperationException("Not implemented");\r
+ }\r
+\r
+ @Override\r
+ public <T> Collection<T> syncRequest(MultiRead<T> request, MultiProcedure<T> procedure) {\r
+ throw new UnsupportedOperationException("Not implemented");\r
+ }\r
+\r
+ @Override\r
+ public <T> void asyncRequest(Read<T> request, Procedure<T> procedure) {\r
+ asyncRequest(request, new NoneToAsyncProcedure<T>(procedure));\r
+ }\r
+\r
+ @Override\r
+ public <T> void asyncRequest(MultiRead<T> request) {\r
+ throw new UnsupportedOperationException("Not implemented");\r
+ }\r
+\r
+ @Override\r
+ public <T> void asyncRequest(MultiRead<T> request, AsyncMultiProcedure<T> procedure) {\r
+ throw new UnsupportedOperationException("Not implemented");\r
+ }\r
+\r
+ @Override\r
+ public <T> void asyncRequest(MultiRead<T> request, MultiProcedure<T> procedure) {\r
+ throw new UnsupportedOperationException("Not implemented");\r
+ }\r
+\r
+ @Override\r
+ public void asyncRequest(Write r) {\r
+ asyncRequest(r, null);\r
+ }\r
+\r
+ @Override\r
+ public void asyncRequest(DelayedWrite r) {\r
+ asyncRequest(r, null);\r
+ }\r
+\r
+ @Override\r
+ public void asyncRequest(WriteOnly r) {\r
+ asyncRequest(r, null);\r
+ }\r
+\r
+ @Override\r
+ public <T> T getService(Class<T> api) {\r
+ return getSession().getService(api);\r
+ }\r
+\r
+ @Override\r
+ public <T> T peekService(Class<T> api) {\r
+ return getSession().peekService(api);\r
+ }\r
+\r
+ @Override\r
+ public boolean hasService(Class<?> api) {\r
+ return getSession().hasService(api);\r
+ }\r
+\r
+ @Override\r
+ public <T> void registerService(Class<T> api, T service) {\r
+ getSession().registerService(api, service);\r
+ }\r
+\r
+// @Override\r
+// public <T> T syncRequest(Read<T> arg0, AsyncListener<T> arg1) {\r
+// throw new UnsupportedOperationException("Not implemented.");\r
+// }\r
+\r
+ @Override\r
+ public <T> T syncRequest(Read<T> arg0, SyncListener<T> arg1) {\r
+ throw new UnsupportedOperationException("Not implemented.");\r
+ }\r
+\r
+ @Override\r
+ public <T> T syncRequest(Read<T> arg0, Listener<T> arg1) {\r
+ throw new UnsupportedOperationException("Not implemented.");\r
+ }\r
+\r
+ @Override\r
+ public <T> T syncRequest(Read<T> arg0, SyncProcedure<T> arg1) {\r
+ throw new UnsupportedOperationException("Not implemented.");\r
+ }\r
+\r
+ @Override\r
+ public <T> T syncRequest(AsyncRead<T> arg0, AsyncListener<T> arg1) {\r
+ throw new UnsupportedOperationException("Not implemented.");\r
+ }\r
+\r
+ @Override\r
+ public <T> T syncRequest(AsyncRead<T> arg0, SyncListener<T> arg1) {\r
+ throw new UnsupportedOperationException("Not implemented.");\r
+\r
+ }\r
+\r
+ @Override\r
+ public <T> T syncRequest(AsyncRead<T> arg0, Listener<T> arg1) {\r
+ throw new UnsupportedOperationException("Not implemented.");\r
+\r
+ }\r
+\r
+ @Override\r
+ public <T> T syncRequest(AsyncRead<T> arg0, SyncProcedure<T> arg1) {\r
+ throw new UnsupportedOperationException("Not implemented.");\r
+\r
+ }\r
+\r
+ @Override\r
+ public <T> void asyncRequest(Read<T> arg0, AsyncListener<T> arg1) {\r
+ throw new UnsupportedOperationException("Not implemented.");\r
+\r
+ }\r
+\r
+ @Override\r
+ public <T> void asyncRequest(Read<T> arg0, SyncListener<T> arg1) {\r
+ throw new UnsupportedOperationException("Not implemented.");\r
+\r
+ }\r
+\r
+ @Override\r
+ public <T> void asyncRequest(Read<T> arg0, Listener<T> arg1) {\r
+ throw new UnsupportedOperationException("Not implemented.");\r
+\r
+ }\r
+\r
+ @Override\r
+ public <T> void asyncRequest(Read<T> arg0, SyncProcedure<T> arg1) {\r
+ throw new UnsupportedOperationException("Not implemented.");\r
+\r
+ }\r
+\r
+ @Override\r
+ public <T> void asyncRequest(AsyncRead<T> arg0, AsyncListener<T> arg1) {\r
+ throw new UnsupportedOperationException("Not implemented.");\r
+\r
+ }\r
+\r
+ @Override\r
+ public <T> void asyncRequest(AsyncRead<T> arg0, SyncListener<T> arg1) {\r
+ throw new UnsupportedOperationException("Not implemented.");\r
+\r
+ }\r
+\r
+ @Override\r
+ public <T> void asyncRequest(AsyncRead<T> arg0, Listener<T> arg1) {\r
+ throw new UnsupportedOperationException("Not implemented.");\r
+\r
+ }\r
+\r
+ @Override\r
+ public <T> void asyncRequest(AsyncRead<T> arg0, SyncProcedure<T> arg1) {\r
+ throw new UnsupportedOperationException("Not implemented.");\r
+\r
+ }\r
+\r
+ @Override\r
+ public <T> Collection<T> syncRequest(MultiRead<T> arg0, AsyncMultiListener<T> arg1) {\r
+ throw new UnsupportedOperationException("Not implemented.");\r
+\r
+ }\r
+\r
+ @Override\r
+ public <T> Collection<T> syncRequest(MultiRead<T> arg0, SyncMultiListener<T> arg1) {\r
+ throw new UnsupportedOperationException("Not implemented.");\r
+\r
+ }\r
+\r
+ @Override\r
+ public <T> Collection<T> syncRequest(MultiRead<T> arg0, MultiListener<T> arg1) {\r
+ throw new UnsupportedOperationException("Not implemented.");\r
+\r
+ }\r
+\r
+ @Override\r
+ public <T> Collection<T> syncRequest(MultiRead<T> arg0, SyncMultiProcedure<T> arg1) {\r
+ throw new UnsupportedOperationException("Not implemented.");\r
+\r
+ }\r
+\r
+ @Override\r
+ public <T> Collection<T> syncRequest(AsyncMultiRead<T> arg0,\r
+ AsyncMultiListener<T> arg1) {\r
+ throw new UnsupportedOperationException("Not implemented.");\r
+\r
+ }\r
+\r
+ @Override\r
+ public <T> Collection<T> syncRequest(AsyncMultiRead<T> arg0, SyncMultiListener<T> arg1) {\r
+ throw new UnsupportedOperationException("Not implemented.");\r
+\r
+ }\r
+\r
+ @Override\r
+ public <T> Collection<T> syncRequest(AsyncMultiRead<T> arg0,\r
+ MultiListener<T> arg1) {\r
+ throw new UnsupportedOperationException("Not implemented.");\r
+\r
+ }\r
+\r
+ @Override\r
+ public <T> Collection<T> syncRequest(AsyncMultiRead<T> arg0, SyncMultiProcedure<T> arg1) {\r
+ throw new UnsupportedOperationException("Not implemented.");\r
+\r
+ }\r
+\r
+ @Override\r
+ public <T> void asyncRequest(MultiRead<T> arg0, AsyncMultiListener<T> arg1) {\r
+ throw new UnsupportedOperationException("Not implemented.");\r
+\r
+ }\r
+\r
+ @Override\r
+ public <T> void asyncRequest(MultiRead<T> arg0, SyncMultiListener<T> arg1) {\r
+ throw new UnsupportedOperationException("Not implemented.");\r
+\r
+ }\r
+\r
+ @Override\r
+ public <T> void asyncRequest(MultiRead<T> arg0, MultiListener<T> arg1) {\r
+ throw new UnsupportedOperationException("Not implemented.");\r
+\r
+ }\r
+\r
+ @Override\r
+ public <T> void asyncRequest(MultiRead<T> arg0, SyncMultiProcedure<T> arg1) {\r
+ throw new UnsupportedOperationException("Not implemented.");\r
+\r
+ }\r
+\r
+ @Override\r
+ public <T> void asyncRequest(AsyncMultiRead<T> arg0,\r
+ AsyncMultiListener<T> arg1) {\r
+ throw new UnsupportedOperationException("Not implemented.");\r
+ }\r
+\r
+ @Override\r
+ public <T> void asyncRequest(AsyncMultiRead<T> arg0, SyncMultiListener<T> arg1) {\r
+ throw new UnsupportedOperationException("Not implemented.");\r
+\r
+ }\r
+\r
+ @Override\r
+ public <T> void asyncRequest(AsyncMultiRead<T> arg0,\r
+ MultiListener<T> arg1) {\r
+ throw new UnsupportedOperationException("Not implemented.");\r
+\r
+ }\r
+\r
+ @Override\r
+ public <T> void asyncRequest(AsyncMultiRead<T> arg0, SyncMultiProcedure<T> arg1) {\r
+ throw new UnsupportedOperationException("Not implemented.");\r
+ }\r
+\r
+ @Override\r
+ public <T> void asyncRequest(ExternalRead<T> request, Procedure<T> procedure) {\r
+ throw new UnsupportedOperationException("Not implemented.");\r
+ }\r
+\r
+ @Override\r
+ public <T> T syncRequest(ExternalRead<T> request) {\r
+ throw new UnsupportedOperationException("Not implemented.");\r
+ }\r
+\r
+ @Override\r
+ public <T> T syncRequest(ExternalRead<T> request, Listener<T> procedure) {\r
+ throw new UnsupportedOperationException("Not implemented.");\r
+ }\r
+\r
+ @Override\r
+ public <T> T syncRequest(ExternalRead<T> request, Procedure<T> procedure) {\r
+ throw new UnsupportedOperationException("Not implemented.");\r
+ }\r
+\r
+ @Override\r
+ public <T> void asyncRequest(ExternalRead<T> request) {\r
+ throw new UnsupportedOperationException("Not implemented.");\r
+ }\r
+\r
+ @Override\r
+ public <T> void asyncRequest(ExternalRead<T> request, Listener<T> procedure) {\r
+ throw new UnsupportedOperationException("Not implemented.");\r
+ }\r
+\r
+ @Override\r
+ public void syncRequest(DelayedWrite request) throws DatabaseException {\r
+ throw new UnsupportedOperationException("Not implemented.");\r
+ }\r
+\r
+ @Override\r
+ public <T> T syncRequest(WriteResult<T> request) throws DatabaseException {\r
+ throw new UnsupportedOperationException();\r
+ }\r
+\r
+ @Override\r
+ public <T> T syncRequest(DelayedWriteResult<T> request)\r
+ throws DatabaseException {\r
+ throw new UnsupportedOperationException();\r
+ }\r
+\r
+ @Override\r
+ public <T> T syncRequest(WriteOnlyResult<T> r) throws DatabaseException {\r
+ throw new UnsupportedOperationException();\r
+ }\r
+\r
+ @Override\r
+ public <T> void asyncRequest(WriteResult<T> r, Procedure<T> procedure) {\r
+ throw new UnsupportedOperationException();\r
+ }\r
+\r
+ @Override\r
+ public <T> void asyncRequest(DelayedWriteResult<T> r, Procedure<T> procedure) {\r
+ throw new UnsupportedOperationException();\r
+ }\r
+\r
+ @Override\r
+ public <T> void asyncRequest(WriteOnlyResult<T> r, Procedure<T> procedure) {\r
+ throw new UnsupportedOperationException();\r
+ }\r
+ \r
+ @Override\r
+ public Resource getRootLibrary() {\r
+ return processor.getRootLibrary();\r
+ }\r
+\r
+ @Override\r
+ public <T> void async(ReadInterface<T> r, Procedure<T> procedure) {\r
+ throw new UnsupportedOperationException();\r
+ }\r
+\r
+ @Override\r
+ public <T> void async(ReadInterface<T> r, AsyncProcedure<T> procedure) {\r
+ throw new UnsupportedOperationException();\r
+ }\r
+\r
+ @Override\r
+ public <T> void async(ReadInterface<T> r, SyncProcedure<T> procedure) {\r
+ throw new UnsupportedOperationException();\r
+ }\r
+\r
+ @Override\r
+ public <T> void async(ReadInterface<T> r, Listener<T> procedure) {\r
+ throw new UnsupportedOperationException();\r
+ }\r
+\r
+ @Override\r
+ public <T> void async(ReadInterface<T> r, AsyncListener<T> procedure) {\r
+ throw new UnsupportedOperationException();\r
+ }\r
+\r
+ @Override\r
+ public <T> void async(ReadInterface<T> r, SyncListener<T> procedure) {\r
+ throw new UnsupportedOperationException();\r
+ }\r
+ \r
+ @Override\r
+ public <T> T sync(ReadInterface<T> r) throws DatabaseException {\r
+ throw new UnsupportedOperationException();\r
+ }\r
+ \r
+ @Override\r
+ public <T> T sync(WriteInterface<T> r) throws DatabaseException {\r
+ throw new UnsupportedOperationException();\r
+ }\r
+ \r
+ @Override\r
+ public <T> void async(WriteInterface<T> r, Procedure<T> procedure) {\r
+ throw new UnsupportedOperationException();\r
+ }\r
+\r
+ @Override\r
+ public <T> void async(WriteInterface<T> r) {\r
+ throw new UnsupportedOperationException();\r
+ }\r
+ \r
+ @Override\r
+ public Object getModificationCounter() {\r
+ throw new UnsupportedOperationException();\r
+ }\r
+ \r
+}\r
Code Review / simantics / platform.git / blobdiff