//
// }
- public <T> Object performQuery(ReadGraphImpl parentGraph, final ExternalRead<T> query, final CacheEntryBase entry_, AsyncProcedure procedure_) throws DatabaseException {
-
- ExternalReadEntry entry = (ExternalReadEntry)entry_;
- AsyncProcedure<T> procedure = (AsyncProcedure<T>)procedure_;
-
- try {
-
- query.register(parentGraph, new Listener<T>() {
-
- AtomicBoolean used = new AtomicBoolean(false);
-
- @Override
- public void execute(T result) {
-
- // Just for safety
- if(entry.isDiscarded()) return;
-
- if(used.compareAndSet(false, true)) {
- //entry.setPending();
- entry.addOrSet(parentGraph.processor, result);
- procedure.execute(parentGraph, result);
- } else {
- entry.queue(result);
- parentGraph.processor.updatePrimitive(query);
- }
-
- }
-
- @Override
- public void exception(Throwable t) {
-
- entry.except(t);
-
- if(used.compareAndSet(false, true)) {
- procedure.exception(parentGraph, t);
- } else {
-// entry.queue(result);
- parentGraph.processor.updatePrimitive(query);
- }
-
- }
-
- @Override
- public String toString() {
- return procedure.toString();
- }
-
- @Override
- public boolean isDisposed() {
- return entry.isDiscarded() || !parentGraph.processor.isBound(entry);
- }
-
- });
-
- return entry.getResult();
-
- } catch (Throwable t) {
-
- entry.except(t);
- procedure.exception(parentGraph, t);
- return entry.getResult();
-
- }
-
- }
public <T> Object performQuery(ReadGraphImpl parentGraph, final AsyncMultiRead<T> query, final CacheEntryBase entry_, Object procedure_) throws DatabaseException {
}
}
- public static void waitPending(QueryProcessor processor, CacheEntry entry) throws DatabaseException {
+ public static void waitPending(ReadGraphImpl graph, CacheEntry entry) throws DatabaseException {
int counter = 0;
while(entry.isPending()) {
try {
- SessionTask task = null;//processor.getOwnTask(processor.thread.get());
- if(task != null) {
- task.run(processor.thread.get());
- } else {
+ boolean performed = graph.performPending();
+ if(!performed) {
Thread.sleep(1);
counter++;
- if(counter > 5000) {
+ if(counter > 30000) {
CacheEntryBase base = ((CacheEntryBase)entry);
// if(base.created != null) {
// System.err.println("created:");
// System.err.println("ready:");
// base.ready.printStackTrace();
// }
- new Exception("Timeout waiting for request to complete: " + entry.getOriginalRequest().toString()).printStackTrace();
- throw new DatabaseException("Timeout waiting for request to complete.");
+ new Exception("Timeout waiting for request to complete: " + entry.getOriginalRequest()).printStackTrace();
+ throw new DatabaseException("Timeout waiting for request to complete." + entry.getOriginalRequest());
//System.err.println("asd");
//base.getQuery().recompute(null, null, entry);
}
}
public static <T> T resultReadEntry(ReadGraphImpl graph, Read r, CacheEntry parent, ListenerBase listener, AsyncProcedure<T> procedure) throws DatabaseException {
- AsyncProcedureWrapper<T> wrap = new AsyncProcedureWrapper<>(procedure);
- QueryCache.runnerReadEntry(graph, r, parent, listener, wrap, true);
- return wrap.get();
+ return (T)QueryCache.runnerReadEntry(graph, r, parent, listener, procedure, true);
}
public static <T> T resultAsyncReadEntry(ReadGraphImpl graph, AsyncRead r, CacheEntry parent, ListenerBase listener, AsyncProcedure<T> procedure) throws DatabaseException {
- AsyncProcedureWrapper<T> wrap = new AsyncProcedureWrapper<>(procedure);
- QueryCache.runnerAsyncReadEntry(graph, r, parent, listener, wrap, true);
- return wrap.get();
+ return (T)QueryCache.runnerAsyncReadEntry(graph, r, parent, listener, procedure, true);
}
public static byte[] resultValueQuery(ReadGraphImpl graph, int r, CacheEntry parent, ListenerBase listener) throws DatabaseException {