import com.google.appengine.api.taskqueue.dev.QueueStateInfo; //导入依赖的package包/类
/**
 * Executes mapreduce tasks, increment the clock between each task.
 *
 * <p>Incrementing the clock between tasks is important if tasks have transactions inside the
 * mapper or reducer, which don't have access to the fake clock.
 *
 * <p>The maxTasks parameter determines how many tasks (at most) will be run. If maxTasks is
 * absent(), all tasks are run until the queue is empty. If maxTasks is zero, no tasks are run.
 */
protected void executeTasks(
    String queueName, @Nullable FakeClock clock, Optional<Integer> maxTasks) throws Exception {
  for (int numTasksDeleted = 0;
      !maxTasks.isPresent() || (numTasksDeleted < maxTasks.get());
      numTasksDeleted++) {
    ofy().clearSessionCache();
    // We have to re-acquire task list every time, because local implementation returns a copy.
    List<QueueStateInfo.TaskStateInfo> taskInfo =
        taskQueue.getQueueStateInfo().get(queueName).getTaskInfo();
    if (taskInfo.isEmpty()) {
      break;
    }
    QueueStateInfo.TaskStateInfo taskStateInfo = taskInfo.get(0);
    taskQueue.deleteTask(queueName, taskStateInfo.getTaskName());
    executeTask(queueName, taskStateInfo);
    if (clock != null) {
      clock.advanceOneMilli();
    }
  }
}
 

import com.google.appengine.api.taskqueue.dev.QueueStateInfo; //导入依赖的package包/类
@Test
public void testExport() throws Exception {
  metrics.projectId = "project id";
  metrics.export();
  QueueStateInfo queueInfo = getLocalTaskQueue().getQueueStateInfo().get("default");
  assertEquals(1, queueInfo.getCountTasks());
  assertEquals("/task/metrics", queueInfo.getTaskInfo().get(0).getUrl());
  Multimap<String, String> params = parseQuery(queueInfo.getTaskInfo().get(0).getBody());
  assertEquals("/path", getOnlyElement(params.get("path")));
  assertEquals("tld", getOnlyElement(params.get("tld")));
  assertEquals("GET", getOnlyElement(params.get("method")));
  assertEquals("0", getOnlyElement(params.get("starttime")));
  assertEquals(
      ImmutableSet.of("foo", "bar"),
      ImmutableSet.copyOf(Splitter.on(' ').split(getOnlyElement(params.get("activity")))));
  assertEquals("456", getOnlyElement(params.get("responsecode")));
  assertTrue(params.containsKey("endtime"));
  assertTrue(params.containsKey("insertid"));
}
 

import com.google.appengine.api.taskqueue.dev.QueueStateInfo; //导入依赖的package包/类
protected void waitUntilTaskQueueIsEmpty() throws InterruptedException {
  boolean hasMoreTasks = true;
  while (hasMoreTasks) {
    Map<String, QueueStateInfo> taskInfoMap = taskQueue.getQueueStateInfo();
    hasMoreTasks = false;
    for (QueueStateInfo taskQueueInfo : taskInfoMap.values()) {
      if (taskQueueInfo.getCountTasks() > 0) {
        hasMoreTasks = true;
        break;
      }
    }
    if (hasMoreTasks) {
      Thread.sleep(100);
    }
  }
}
 

import com.google.appengine.api.taskqueue.dev.QueueStateInfo; //导入依赖的package包/类
/**
 * Ensures that the only tasks in the named queue are exactly those that match the expected
 * matchers.
 */
public static void assertTasksEnqueued(String queueName, Collection<TaskMatcher> taskMatchers)
    throws Exception {
  QueueStateInfo qsi = getQueueInfo(queueName);
  assertThat(qsi.getTaskInfo()).hasSize(taskMatchers.size());
  List<TaskStateInfo> taskInfos = new ArrayList<>(qsi.getTaskInfo());
  for (final TaskMatcher taskMatcher : taskMatchers) {
    try {
      taskInfos.remove(taskInfos.stream().filter(taskMatcher).findFirst().get());
    } catch (NoSuchElementException e) {
      final Map<String, Object> taskMatcherMap = taskMatcher.expected.toMap();
      assert_()
          .fail(
              "Task not found in queue %s:\n\n%s\n\nPotential candidate match diffs:\n\n%s",
              queueName,
              taskMatcher,
              taskInfos
                  .stream()
                  .map(
                      input ->
                          prettyPrintEntityDeepDiff(
                              taskMatcherMap,
                              Maps.filterKeys(
                                  new MatchableTaskInfo(input).toMap(),
                                  in(taskMatcherMap.keySet()))))
                  .collect(joining("\n")));
    }
  }
}
 

import com.google.appengine.api.taskqueue.dev.QueueStateInfo; //导入依赖的package包/类
private void doTest() throws InterruptedException {
  QueueFactory.getDefaultQueue().add(TaskOptions.Builder.withTaskName("task29"));
  // Give the task time to execute if tasks are actually enabled (which they
  // aren't, but that's part of the test).
  Thread.sleep(1000);
  LocalTaskQueue ltq = LocalTaskQueueTestConfig.getLocalTaskQueue();
  QueueStateInfo qsi = ltq.getQueueStateInfo().get(QueueFactory.getDefaultQueue().getQueueName());
  assertEquals(1, qsi.getTaskInfo().size());
  assertEquals("task29", qsi.getTaskInfo().get(0).getTaskName());
}
 

import com.google.appengine.api.taskqueue.dev.QueueStateInfo; //导入依赖的package包/类
private void doTest() throws InterruptedException {
  // [START QueueFactory]
  QueueFactory.getQueue("my-queue-name").add(TaskOptions.Builder.withTaskName("task29"));
  // [END QueueFactory]
  // Give the task time to execute if tasks are actually enabled (which they
  // aren't, but that's part of the test).
  Thread.sleep(1000);
  LocalTaskQueue ltq = LocalTaskQueueTestConfig.getLocalTaskQueue();
  QueueStateInfo qsi =
      ltq.getQueueStateInfo().get(QueueFactory.getQueue("my-queue-name").getQueueName());
  assertEquals(1, qsi.getTaskInfo().size());
  assertEquals("task29", qsi.getTaskInfo().get(0).getTaskName());
}
 

import com.google.appengine.api.taskqueue.dev.QueueStateInfo; //导入依赖的package包/类
/**
 * Asserts that the {@code numExpectedTasksInQueue} matches the actual number of tasks in the queue.
 */
private void assertNumTasksInQueue(int numExpectedTasksInQueue)
{
	LocalTaskQueue ltq = LocalTaskQueueTestConfig.getLocalTaskQueue();
	QueueStateInfo qsi = ltq.getQueueStateInfo()
		.get(QueueFactory.getQueue(DELETE_COUNTER_SHARD_QUEUE_NAME).getQueueName());
	assertEquals(numExpectedTasksInQueue, qsi.getTaskInfo().size());
}
 

import com.google.appengine.api.taskqueue.dev.QueueStateInfo; //导入依赖的package包/类
/** Returns the info object for the provided queue name. */
public static QueueStateInfo getQueueInfo(String queueName) {
  return getLocalTaskQueue().getQueueStateInfo().get(queueName);
}
 

import com.google.appengine.api.taskqueue.dev.QueueStateInfo; //导入依赖的package包/类
protected List<QueueStateInfo.TaskStateInfo> getTasks(String queueName) {
  return taskQueue.getQueueStateInfo().get(queueName).getTaskInfo();
}
 

import com.google.appengine.api.taskqueue.dev.QueueStateInfo; //导入依赖的package包/类
protected void executeTask(String queueName, QueueStateInfo.TaskStateInfo taskStateInfo)
    throws Exception {
  logger.finefmt("Executing task %s with URL %s",
      taskStateInfo.getTaskName(), taskStateInfo.getUrl());
  // Hack to allow for deferred tasks. Exploits knowing how they work.
  if (taskStateInfo.getUrl().endsWith("__deferred__")) {
    ObjectInputStream oin =
        new ObjectInputStream(new ByteArrayInputStream(taskStateInfo.getBodyAsBytes()));
    Runnable object = (Runnable) oin.readObject();
    object.run();
    return;
  }
  HttpServletRequest request = mock(HttpServletRequest.class);
  HttpServletResponse response = mock(HttpServletResponse.class);

  // Strip off routing paths that are handled in web.xml in non-test scenarios.
  String pathInfo = taskStateInfo.getUrl();
  if (pathInfo.startsWith("/_dr/mapreduce/")) {
    pathInfo = pathInfo.replace("/_dr/mapreduce", "");
  } else if (pathInfo.startsWith("/mapreduce/")) {
      pathInfo = pathInfo.replace("/mapreduce", "");
  } else if (pathInfo.startsWith("/")) {
    pathInfo = pathInfo.replace("/_ah/", "");
    pathInfo = pathInfo.substring(pathInfo.indexOf('/'));
  } else {
    pathInfo = "/" + pathInfo;
  }
  when(request.getPathInfo()).thenReturn(pathInfo);
  when(request.getHeader("X-AppEngine-QueueName")).thenReturn(queueName);
  when(request.getHeader("X-AppEngine-TaskName")).thenReturn(taskStateInfo.getTaskName());
  // Pipeline looks at this header but uses the value only for diagnostic messages
  when(request.getIntHeader(TaskHandler.TASK_RETRY_COUNT_HEADER)).thenReturn(-1);
  for (HeaderWrapper header : taskStateInfo.getHeaders()) {
    int value = parseAsQuotedInt(header.getValue());
    when(request.getIntHeader(header.getKey())).thenReturn(value);
    logger.finefmt("header: %s=%s", header.getKey(), header.getValue());
    when(request.getHeader(header.getKey())).thenReturn(header.getValue());
  }

  Map<String, String> parameters = decodeParameters(taskStateInfo.getBody());
  for (String name : parameters.keySet()) {
    when(request.getParameter(name)).thenReturn(parameters.get(name));
  }
  when(request.getParameterNames()).thenReturn(Collections.enumeration(parameters.keySet()));

  if (taskStateInfo.getMethod().equals("POST")) {
    if (taskStateInfo.getUrl().startsWith(PipelineServlet.BASE_URL)) {
      pipelineServlet.doPost(request, response);
    } else {
      mrServlet.doPost(request, response);
    }
  } else {
    throw new UnsupportedOperationException();
  }
}
 

import com.google.appengine.api.taskqueue.dev.QueueStateInfo; //导入依赖的package包/类
private static int getCountTasks() {
    LocalTaskQueue localTaskQueue = LocalTaskQueueTestConfig.getLocalTaskQueue();

    Map<String, QueueStateInfo> queueStateInfo = localTaskQueue.getQueueStateInfo();

    int count = 0;

    for (String key : queueStateInfo.keySet()) {
        count += queueStateInfo.get(key).getCountTasks();
    }

    return count;
}
 

import com.google.appengine.api.taskqueue.dev.QueueStateInfo; //导入依赖的package包/类
/**
 * Fetches the number of tasks that have been scheduled in a named queue in the
 * current test, including both tasks that have executed, and those that have not.
 * @param queueName the name of the queue
 * @return the total number of tasks enqueued by the test
 */
public int getTaskCount(String queueName) {
    LocalTaskQueue localTaskQueue = LocalTaskQueueTestConfig.getLocalTaskQueue();
    QueueStateInfo queueState = localTaskQueue.getQueueStateInfo().get(queueName);
    return queueState.getCountTasks();
}