@Test public void noActiveTaskSource() throws Exception { for (QueryTaskSource x : sourceAggregator.taskSources) { when(x.next()).thenReturn(null); when(x.complete()).thenReturn(false); when(x.hasNoActiveSources()).thenReturn(true); } queryTask.run(); verify(sourceAggregator, times(4)).tryActivateSource(any(QueryTaskSource.class)); verify(sourceAggregator.executor).execute(queryTask); }
@Override public void run() { try { if (sourceAggregator.queryPromise.isDone()) { return; } if (!sourceAggregator.channelWritable) { sourceAggregator.needScheduling = true; return; } List<QueryTaskSource> taskSources = queryTaskSourceSupplier.get(); int bundlesProcessed = readBundles(taskSources, AggregateConfig.FRAME_READER_READS); if (bundlesProcessed > 0) { sourceAggregator.queryPromise.tryProgress(0, bundlesProcessed); } if (sourceAggregator.completed == sourceAggregator.totalTasks) { if (!sourceAggregator.queryPromise.trySuccess()) { log.warn("Tried to complete queryPromise {} , but failed", sourceAggregator.queryPromise); } } else { if (bundlesProcessed > 0) { pollFailures = 0; sourceAggregator.executor.execute(this); } else { pollFailures += 1; rescheduleSelfWithBackoff(); } } } catch (Throwable e) { if (!sourceAggregator.queryPromise.tryFailure(e)) { log.warn("Tried to fail queryPromise {} , but failed", sourceAggregator.queryPromise, e); } } }
QueryTask implements Runnable { @Override public void run() { try { if (sourceAggregator.queryPromise.isDone()) { return; } if (!sourceAggregator.channelWritable) { sourceAggregator.needScheduling = true; return; } List<QueryTaskSource> taskSources = queryTaskSourceSupplier.get(); int bundlesProcessed = readBundles(taskSources, AggregateConfig.FRAME_READER_READS); if (bundlesProcessed > 0) { sourceAggregator.queryPromise.tryProgress(0, bundlesProcessed); } if (sourceAggregator.completed == sourceAggregator.totalTasks) { if (!sourceAggregator.queryPromise.trySuccess()) { log.warn("Tried to complete queryPromise {} , but failed", sourceAggregator.queryPromise); } } else { if (bundlesProcessed > 0) { pollFailures = 0; sourceAggregator.executor.execute(this); } else { pollFailures += 1; rescheduleSelfWithBackoff(); } } } catch (Throwable e) { if (!sourceAggregator.queryPromise.tryFailure(e)) { log.warn("Tried to fail queryPromise {} , but failed", sourceAggregator.queryPromise, e); } } } }
QueryTask implements Runnable { @Override public void run() { try { if (sourceAggregator.queryPromise.isDone()) { return; } if (!sourceAggregator.channelWritable) { sourceAggregator.needScheduling = true; return; } List<QueryTaskSource> taskSources = queryTaskSourceSupplier.get(); int bundlesProcessed = readBundles(taskSources, AggregateConfig.FRAME_READER_READS); if (bundlesProcessed > 0) { sourceAggregator.queryPromise.tryProgress(0, bundlesProcessed); } if (sourceAggregator.completed == sourceAggregator.totalTasks) { if (!sourceAggregator.queryPromise.trySuccess()) { log.warn("Tried to complete queryPromise {} , but failed", sourceAggregator.queryPromise); } } else { if (bundlesProcessed > 0) { pollFailures = 0; sourceAggregator.executor.execute(this); } else { pollFailures += 1; rescheduleSelfWithBackoff(); } } } catch (Throwable e) { if (!sourceAggregator.queryPromise.tryFailure(e)) { log.warn("Tried to fail queryPromise {} , but failed", sourceAggregator.queryPromise, e); } } } QueryTask(MeshSourceAggregator sourceAggregator); }
QueryTask implements Runnable { @Override public void run() { try { if (sourceAggregator.queryPromise.isDone()) { return; } if (!sourceAggregator.channelWritable) { sourceAggregator.needScheduling = true; return; } List<QueryTaskSource> taskSources = queryTaskSourceSupplier.get(); int bundlesProcessed = readBundles(taskSources, AggregateConfig.FRAME_READER_READS); if (bundlesProcessed > 0) { sourceAggregator.queryPromise.tryProgress(0, bundlesProcessed); } if (sourceAggregator.completed == sourceAggregator.totalTasks) { if (!sourceAggregator.queryPromise.trySuccess()) { log.warn("Tried to complete queryPromise {} , but failed", sourceAggregator.queryPromise); } } else { if (bundlesProcessed > 0) { pollFailures = 0; sourceAggregator.executor.execute(this); } else { pollFailures += 1; rescheduleSelfWithBackoff(); } } } catch (Throwable e) { if (!sourceAggregator.queryPromise.tryFailure(e)) { log.warn("Tried to fail queryPromise {} , but failed", sourceAggregator.queryPromise, e); } } } QueryTask(MeshSourceAggregator sourceAggregator); @Override void run(); }
QueryTask implements Runnable { @Override public void run() { try { if (sourceAggregator.queryPromise.isDone()) { return; } if (!sourceAggregator.channelWritable) { sourceAggregator.needScheduling = true; return; } List<QueryTaskSource> taskSources = queryTaskSourceSupplier.get(); int bundlesProcessed = readBundles(taskSources, AggregateConfig.FRAME_READER_READS); if (bundlesProcessed > 0) { sourceAggregator.queryPromise.tryProgress(0, bundlesProcessed); } if (sourceAggregator.completed == sourceAggregator.totalTasks) { if (!sourceAggregator.queryPromise.trySuccess()) { log.warn("Tried to complete queryPromise {} , but failed", sourceAggregator.queryPromise); } } else { if (bundlesProcessed > 0) { pollFailures = 0; sourceAggregator.executor.execute(this); } else { pollFailures += 1; rescheduleSelfWithBackoff(); } } } catch (Throwable e) { if (!sourceAggregator.queryPromise.tryFailure(e)) { log.warn("Tried to fail queryPromise {} , but failed", sourceAggregator.queryPromise, e); } } } QueryTask(MeshSourceAggregator sourceAggregator); @Override void run();  }