ResultCompletePolicy implements IResultPolicy<T> { public long waitTime(ClientResult<T> result) { boolean done = result.isClosed(); long now = System.currentTimeMillis(); if (!done) { if (now < completeStopTime) { done = result.isComplete(); } else if (now < coverageStopTime) { done = result.getShardCoverage() >= coverage; } else { done = true; } } if (done) { return shutDown ? -1 : 0; } return coverageStopTime - now; } ResultCompletePolicy(long timeout);  ResultCompletePolicy(long timeout, boolean shutDown);  ResultCompletePolicy(long completeWait, long coverageWait, double coverage, boolean shutDown); long waitTime(ClientResult<T> result); @Override String toString();  }
@Test public void testCompleteTiming() { ClientResult<String> r = new ClientResult<String>(null, "a", "b", "c"); ResultCompletePolicy<String> rc = new ResultCompletePolicy<String>(1000, true); long now = System.currentTimeMillis(); long start = System.currentTimeMillis(); long stop = start + 500; while (now < stop) { assertTrue(rc.waitTime(r) > 400); sleep(1); now = System.currentTimeMillis(); } stop = start + 1500; while (now < stop) { sleep(stop - now); now = System.currentTimeMillis(); } stop = start + 2000; while (now < stop) { assertTrue(rc.waitTime(r) == -1); sleep(1); now = System.currentTimeMillis(); } }
@Test public void testCoverage() { Set<String> shards = new HashSet<String>(); for (int i = 0; i < 1000; i++) { shards.add("s" + i); } ClientResult<String> r = new ClientResult<String>(null, shards); ResultCompletePolicy<String> rc = new ResultCompletePolicy<String>(0, 60000, (879.0 / 1000.0), false); for (int i = 0; i < 1000; i++) { try { r.addResult("foo", "s" + i); if (i < 878) { assertTrue(rc.waitTime(r) > 30000); } else { assertTrue(rc.waitTime(r) == 0); } } catch (Error e) { System.err.println("i = " + i); throw e; } } }