WorkQueue implements INodeExecutor { protected WorkQueue(INodeProxyManager shardManager, Set<String> allShards, Method method, int shardArrayParamIndex, Object... args) { this(new INodeInteractionFactory<T>() { public Runnable createInteraction(Method method, Object[] args, int shardArrayParamIndex, String node, Map<String, List<String>> nodeShardMap, int tryCount, int maxTryCount, INodeProxyManager shardManager, INodeExecutor nodeExecutor, IResultReceiver<T> results) { return new NodeInteraction<T>(method, args, shardArrayParamIndex, node, nodeShardMap, tryCount, maxTryCount, shardManager, nodeExecutor, results); } }, shardManager, allShards, method, shardArrayParamIndex, args); } protected  WorkQueue(INodeProxyManager shardManager, Set<String> allShards, Method method, int shardArrayParamIndex,
          Object... args); protected  WorkQueue(INodeInteractionFactory<T> interactionFactory, INodeProxyManager shardManager,
          Set<String> allShards, Method method, int shardArrayParamIndex, Object... args); static void resetInstanceCounter(); void execute(String node, Map<String, List<String>> nodeShardMap, int tryCount, int maxTryCount); void shutdown(); ClientResult<T> getResults(long timeout); ClientResult<T> getResults(long timeout, boolean shutdown); ClientResult<T> getResults(IResultPolicy<T> policy); @Override String toString();  }
@Test public void testWorkQueue() throws Exception { TestShardManager sm = new TestShardManager(); Method method = TestServer.class.getMethod("doSomething", Integer.TYPE); WorkQueue.resetInstanceCounter(); for (int i = 0; i < 500; i++) { sm.reset(); TestNodeInteractionFactory factory = new TestNodeInteractionFactory(10); WorkQueue<Integer> wq = new WorkQueue<Integer>(factory, sm, sm.allShards(), method, -1, 16); assertEquals(String.format("WorkQueue[TestServer.doSomething(16) (id=%d)]", i), wq.toString()); Map<String, List<String>> plan = sm.createNode2ShardsMap(sm.allShards()); for (String node : plan.keySet()) { wq.execute(node, plan, 1, 3); } ClientResult<Integer> r = wq.getResults(1000); int numNodes = plan.keySet().size(); int numShards = sm.allShards().size(); assertEquals(String.format("ClientResult: %d results, 0 errors, %d/%d shards (closed) (complete)", numNodes, numShards, numShards), r.toString()); assertEquals(6, factory.getCalls().size()); wq.shutdown(); } }