@Test public void testGetResultOfQueuedTaskAfterItExecutes() throws Exception { final CountDownLatch latch = new CountDownLatch(1); em.submit(MutableMap.of("tag", "category1"), newLatchAwaiter(latch)); BasicTask<Integer> t = new BasicTask<Integer>(Callables.returning(123)); Future<Integer> future = em.submit(MutableMap.of("tag", "category1"), t); latch.countDown(); assertEquals(future.get(), (Integer)123); }
@Override public synchronized <T> Future<T> submit(Callable<T> c) { if (running.compareAndSet(false, true)) { return executeNow(c); } else { WrappingFuture<T> f = new WrappingFuture<T>(); order.add(new QueuedSubmission<T>(c, f)); queueSize++; if (queueSize>0 && (queueSize == 50 || (queueSize<=500 && (queueSize%100)==0) || (queueSize%1000)==0) && queueSize!=lastSizeWarn) { LOG.warn("{} is backing up, {} tasks queued", this, queueSize); if (LOG.isDebugEnabled()) { LOG.debug("Task queue backing up detail, queue "+this+"; task context is "+Tasks.current()+"; latest task is "+c+"; first task is "+order.peek()); } lastSizeWarn = queueSize; } return f; } }
SingleThreadedScheduler implements TaskScheduler, CanSetName { @Override public synchronized <T> Future<T> submit(Callable<T> c) { if (running.compareAndSet(false, true)) { return executeNow(c); } else { WrappingFuture<T> f = new WrappingFuture<T>(); order.add(new QueuedSubmission<T>(c, f)); queueSize++; if (queueSize>0 && (queueSize == 50 || (queueSize<=500 && (queueSize%100)==0) || (queueSize%1000)==0) && queueSize!=lastSizeWarn) { LOG.warn("{} is backing up, {} tasks queued", this, queueSize); if (LOG.isDebugEnabled()) { LOG.debug("Task queue backing up detail, queue "+this+"; task context is "+Tasks.current()+"; latest task is "+c+"; first task is "+order.peek()); } lastSizeWarn = queueSize; } return f; } } }
SingleThreadedScheduler implements TaskScheduler, CanSetName { @Override public synchronized <T> Future<T> submit(Callable<T> c) { if (running.compareAndSet(false, true)) { return executeNow(c); } else { WrappingFuture<T> f = new WrappingFuture<T>(); order.add(new QueuedSubmission<T>(c, f)); queueSize++; if (queueSize>0 && (queueSize == 50 || (queueSize<=500 && (queueSize%100)==0) || (queueSize%1000)==0) && queueSize!=lastSizeWarn) { LOG.warn("{} is backing up, {} tasks queued", this, queueSize); if (LOG.isDebugEnabled()) { LOG.debug("Task queue backing up detail, queue "+this+"; task context is "+Tasks.current()+"; latest task is "+c+"; first task is "+order.peek()); } lastSizeWarn = queueSize; } return f; } }  }
SingleThreadedScheduler implements TaskScheduler, CanSetName { @Override public synchronized <T> Future<T> submit(Callable<T> c) { if (running.compareAndSet(false, true)) { return executeNow(c); } else { WrappingFuture<T> f = new WrappingFuture<T>(); order.add(new QueuedSubmission<T>(c, f)); queueSize++; if (queueSize>0 && (queueSize == 50 || (queueSize<=500 && (queueSize%100)==0) || (queueSize%1000)==0) && queueSize!=lastSizeWarn) { LOG.warn("{} is backing up, {} tasks queued", this, queueSize); if (LOG.isDebugEnabled()) { LOG.debug("Task queue backing up detail, queue "+this+"; task context is "+Tasks.current()+"; latest task is "+c+"; first task is "+order.peek()); } lastSizeWarn = queueSize; } return f; } }  @Override void setName(String name); @Override String toString(); @Override void injectExecutor(ExecutorService executor); @Override synchronized Future<T> submit(Callable<T> c); }
SingleThreadedScheduler implements TaskScheduler, CanSetName { @Override public synchronized <T> Future<T> submit(Callable<T> c) { if (running.compareAndSet(false, true)) { return executeNow(c); } else { WrappingFuture<T> f = new WrappingFuture<T>(); order.add(new QueuedSubmission<T>(c, f)); queueSize++; if (queueSize>0 && (queueSize == 50 || (queueSize<=500 && (queueSize%100)==0) || (queueSize%1000)==0) && queueSize!=lastSizeWarn) { LOG.warn("{} is backing up, {} tasks queued", this, queueSize); if (LOG.isDebugEnabled()) { LOG.debug("Task queue backing up detail, queue "+this+"; task context is "+Tasks.current()+"; latest task is "+c+"; first task is "+order.peek()); } lastSizeWarn = queueSize; } return f; } }  @Override void setName(String name); @Override String toString(); @Override void injectExecutor(ExecutorService executor); @Override synchronized Future<T> submit(Callable<T> c);  }