import com.google.appengine.api.taskqueue.TaskHandle; //导入依赖的package包/类
/** Leases and returns all tasks from the queue with the specified tag tld, in batches. */
public static List<TaskHandle> loadAllTasks(Queue queue, String tld) {
  ImmutableList.Builder<TaskHandle> allTasks = new ImmutableList.Builder<>();
  int numErrors = 0;
  long backOff = backOffMillis;
  while (true) {
    try {
      List<TaskHandle> tasks = queue.leaseTasks(LeaseOptions.Builder
          .withTag(tld)
          .leasePeriod(LEASE_PERIOD.getMillis(), TimeUnit.MILLISECONDS)
          .countLimit(BATCH_SIZE));
      allTasks.addAll(tasks);
      if (tasks.isEmpty()) {
        return allTasks.build();
      }
    } catch (TransientFailureException | DeadlineExceededException e) {
      if (++numErrors >= 3) {
        throw new RuntimeException("Error leasing tasks", e);
      }
      Uninterruptibles.sleepUninterruptibly(backOff, TimeUnit.MILLISECONDS);
      backOff *= 2;
    }
  }
}
 

import com.google.appengine.api.taskqueue.TaskHandle; //导入依赖的package包/类
/** Deletes a list of tasks from the given queue using a retrier. */
private static void deleteTasksWithRetry(
    final List<DnsRefreshRequest> refreshRequests,
    final Queue queue,
    AsyncFlowMetrics asyncFlowMetrics,
    Retrier retrier,
    OperationResult result) {
  if (refreshRequests.isEmpty()) {
    return;
  }
  final List<TaskHandle> tasks =
      refreshRequests.stream().map(DnsRefreshRequest::task).collect(toImmutableList());
  retrier.callWithRetry(() -> queue.deleteTask(tasks), TransientFailureException.class);
  refreshRequests.forEach(
      r -> asyncFlowMetrics.recordAsyncFlowResult(DNS_REFRESH, result, r.requestedTime()));
}
 

import com.google.appengine.api.taskqueue.TaskHandle; //导入依赖的package包/类
/**
 * Returns a packaged-up {@link DnsRefreshRequest} parsed from a task queue task.
 */
static DnsRefreshRequest createFromTask(TaskHandle task, DateTime now) throws Exception {
  ImmutableMap<String, String> params = ImmutableMap.copyOf(task.extractParams());
  Key<HostResource> hostKey =
      Key.create(checkNotNull(params.get(PARAM_HOST_KEY), "Host to refresh not specified"));
  HostResource host =
      checkNotNull(ofy().load().key(hostKey).now(), "Host to refresh doesn't exist");
  boolean isHostDeleted =
      isDeleted(host, latestOf(now, host.getUpdateAutoTimestamp().getTimestamp()));
  if (isHostDeleted) {
    logger.infofmt("Host %s is already deleted, not refreshing DNS.", hostKey);
  }
  return new AutoValue_RefreshDnsOnHostRenameAction_DnsRefreshRequest.Builder()
      .setHostKey(hostKey)
      .setLastUpdateTime(host.getUpdateAutoTimestamp().getTimestamp())
      .setRequestedTime(
          DateTime.parse(
              checkNotNull(params.get(PARAM_REQUESTED_TIME), "Requested time not specified")))
      .setIsRefreshNeeded(!isHostDeleted)
      .setTask(task)
      .build();
}
 

import com.google.appengine.api.taskqueue.TaskHandle; //导入依赖的package包/类
private static String processTasks(List<TaskHandle> tasks, Queue q) {
  String payload;
  int numberOfDeletedTasks = 0;
  for (TaskHandle task : tasks) {
    payload = new String(task.getPayload());
    output = String.format("Processing: taskName='%s'  payload='%s'", task.getName(), payload);
    log.info(output);
    output = String.format("Deleting taskName='%s'", task.getName());
    log.info(output);
    // [START delete_task]
    q.deleteTask(task);
    // [END delete_task]
    numberOfDeletedTasks++;
  }
  if (numberOfDeletedTasks > 0) {
    message = "Processed and deleted " + numberOfTasksToLease + " tasks from the "
        + " task queue.";
  } else {
    message = "Task Queue has no tasks available for lease.";
  }
  return message;
}
 

import com.google.appengine.api.taskqueue.TaskHandle; //导入依赖的package包/类
private static String processTasks(List<TaskHandle> tasks, Queue q) {
  String payload;
  int numberOfDeletedTasks = 0;
  for (TaskHandle task : tasks) {
    payload = new String(task.getPayload());
    output =
        String.format(
            "Processing: taskName='%s'  payload='%s'",
            task.getName().toString(), payload.toString());
    log.info(output.toString());
    output = String.format("Deleting taskName='%s'", task.getName().toString());
    log.info(output.toString());
    // [START delete_task]
    q.deleteTask(task);
    // [END delete_task]
    numberOfDeletedTasks++;
  }
  if (numberOfDeletedTasks > 0) {
    message =
        "Processed and deleted " + numberOfTasksToLease + " tasks from the " + " task queue.";
  } else {
    message = "Task Queue has no tasks available for lease.";
  }
  return message;
}
 

import com.google.appengine.api.taskqueue.TaskHandle; //导入依赖的package包/类
List<TaskHandle> addToQueue(final Collection<Task> tasks) {
  List<TaskHandle> handles = new ArrayList<>();
  Map<String, List<TaskOptions>> queueNameToTaskOptions = new HashMap<>();
  for (Task task : tasks) {
    logger.finest("Enqueueing: " + task);
    String queueName = task.getQueueSettings().getOnQueue();
    TaskOptions taskOptions = toTaskOptions(task);
    List<TaskOptions> taskOptionsList = queueNameToTaskOptions.get(queueName);
    if (taskOptionsList == null) {
      taskOptionsList = new ArrayList<>();
      queueNameToTaskOptions.put(queueName, taskOptionsList);
    }
    taskOptionsList.add(taskOptions);
  }
  for (Map.Entry<String, List<TaskOptions>> entry : queueNameToTaskOptions.entrySet()) {
    Queue queue = getQueue(entry.getKey());
    handles.addAll(addToQueue(queue, entry.getValue()));
  }
  return handles;
}
 

import com.google.appengine.api.taskqueue.TaskHandle; //导入依赖的package包/类
public void testEnqueueBatchTasks() {
  AppEngineTaskQueue queue = new AppEngineTaskQueue();
  List<Task> tasks = new ArrayList<>(AppEngineTaskQueue.MAX_TASKS_PER_ENQUEUE);
  for (int i = 0; i < AppEngineTaskQueue.MAX_TASKS_PER_ENQUEUE; i++) {
    Task task = createTask();
    tasks.add(task);
  }
  List<TaskHandle> handles = queue.addToQueue(tasks);
  assertEquals(AppEngineTaskQueue.MAX_TASKS_PER_ENQUEUE, handles.size());
  for (int i = 0; i < AppEngineTaskQueue.MAX_TASKS_PER_ENQUEUE; i++) {
    assertEquals(tasks.get(i).getName(), handles.get(i).getName());
  }

  handles = queue.addToQueue(tasks);
  assertEquals(0, handles.size());
}
 

import com.google.appengine.api.taskqueue.TaskHandle; //导入依赖的package包/类
public void testEnqueueLargeBatchTasks() {
  AppEngineTaskQueue queue = new AppEngineTaskQueue();
  int batchSize = AppEngineTaskQueue.MAX_TASKS_PER_ENQUEUE * 2 + 10;
  List<Task> tasks = new ArrayList<>(batchSize);
  for (int i = 0; i < batchSize; i++) {
    Task task = createTask();
    tasks.add(task);
  }
  List<TaskHandle> handles = queue.addToQueue(tasks);
  assertEquals(tasks.size(), handles.size());
  for (int i = 0; i < tasks.size(); i++) {
    assertEquals(tasks.get(i).getName(), handles.get(i).getName());
  }

  handles = queue.addToQueue(tasks);
  assertEquals(0, handles.size());
}
 

import com.google.appengine.api.taskqueue.TaskHandle; //导入依赖的package包/类
public void testEnqueueBatchTwoStages() {
  AppEngineTaskQueue queue = new AppEngineTaskQueue();
  int batchSize = AppEngineTaskQueue.MAX_TASKS_PER_ENQUEUE * 2;
  List<Task> tasks = new ArrayList<>(batchSize);
  for (int i = 0; i < batchSize; i++) {
    Task task = createTask();
    tasks.add(task);
  }

  int firstBatchSize = AppEngineTaskQueue.MAX_TASKS_PER_ENQUEUE;
  List<TaskHandle> handles = queue.addToQueue(tasks.subList(0, firstBatchSize));

  assertEquals(firstBatchSize, handles.size());
  for (int i = 0; i < firstBatchSize; i++) {
    assertEquals(tasks.get(i).getName(), handles.get(i).getName());
  }

  handles = queue.addToQueue(tasks);

  // Duplicate is rejected (not counted) per batch.
  int expected = tasks.size() - firstBatchSize;
  assertEquals(expected, handles.size());
  for (int i = 0; i < expected; i++) {
    assertEquals(tasks.get(firstBatchSize + i).getName(), handles.get(i).getName());
  }
}
 

import com.google.appengine.api.taskqueue.TaskHandle; //导入依赖的package包/类
private static void deferredTask(HttpServletRequest req, HttpServletResponse resp)
    throws IOException {
  String queue = req.getParameter("queue");
  Queue q;
  if (queue == null) {
    q = QueueFactory.getDefaultQueue();
  } else {
    q = QueueFactory.getQueue(queue);
  }
  final String data = req.getParameter("deferredData");

  TaskOptions opts =
      TaskOptions.Builder.withPayload(
          new DeferredTask() {
            @Override
            public void run() {
              gotCalledBack(data);
            }
          });

  latch = new CountDownLatch(1);
  TaskHandle handle = q.add(opts);
  resp.getWriter().print(handle.getQueueName());
}
 

import com.google.appengine.api.taskqueue.TaskHandle; //导入依赖的package包/类
private void leaseAndDeleteTasks(HttpServletRequest req, HttpServletResponse resp)
    throws ServletException, IOException {
  int numTasks = Integer.parseInt(req.getParameter("numTasks"));
  Double lease = Double.parseDouble(req.getParameter("lease"));
  String queue = req.getParameter("queue");
  Queue q = QueueFactory.getQueue(queue);
  Boolean doDelete = Boolean.parseBoolean(req.getParameter("doDelete"));

  List<TaskHandle> tasks = q.leaseTasks(lease.intValue() * 1000, TimeUnit.MILLISECONDS, numTasks);

  for (TaskHandle task : tasks) {
    if (doDelete) {
      q.deleteTask(task.getName());
    }
  }
  resp.getWriter().print(queue + "," + tasks.size());
}
 

import com.google.appengine.api.taskqueue.TaskHandle; //导入依赖的package包/类
@Test
public void testBasicLease() {
    String taskGroupTag = "testBasicLease";
    String taskBaseName = taskGroupTag + "_" + getTimeStampRandom();
    int count = 10;

    addTasks(count, taskBaseName, taskGroupTag, payload);

    // lease 1 task from pull queue
    List<TaskHandle> handleList = leaseTasksByTag60Secs(taskGroupTag, 1, false);
    assertEquals(1, handleList.size());
    deleteTaskByName(handleList.get(0).getName());

    // lease the rest from pull queue
    handleList = leaseTasksByTag60Secs(taskGroupTag, MAX_LEASE_COUNT, false);
    assertEquals(count - 1, handleList.size());

    for (TaskHandle t : handleList) {
        assertTrue(t.getName().startsWith(taskBaseName));
        assertEquals(E2E_TESTING_PULL, t.getQueueName());
        assertEquals(payload, new String(t.getPayload()));
        deleteTaskByName(t.getName());
    }
}
 

import com.google.appengine.api.taskqueue.TaskHandle; //导入依赖的package包/类
@Test
public void testLeaseExpiration() {
    String groupTag = "testLeaseExpiration";
    String taskBaseName = groupTag + "_" + getTimeStampRandom();
    deleteOnTearDownTags.add(taskBaseName);

    int count = 1;
    List<TaskHandle> taskList = addTasks(count, taskBaseName, groupTag, "nada");
    long leaseDuration = 1000;
    LeaseOptions options = LeaseOptions.Builder
        .withTag(groupTag)
        .countLimit(count)
        .leasePeriod(leaseDuration, TimeUnit.MILLISECONDS);

    List<TaskHandle> tasks = queue.leaseTasks(options);
    assertEquals(count, tasks.size());

    sync(leaseDuration + 1000); // wait for lease to expire
    List<TaskHandle> tasksAfterExpire = queue.leaseTasks(options);

    // expired, so it should be available for lease.
    assertEquals(count, tasksAfterExpire.size());

    deleteMultipleTasks(tasksAfterExpire);
}
 

import com.google.appengine.api.taskqueue.TaskHandle; //导入依赖的package包/类
@Test
public void testLeaseTasksByTagBytes() {
    String groupTag = "testLeaseTasksByTagBytes";
    String taskBaseName = groupTag + "_" + getTimeStampRandom();
    taskTags.add(taskBaseName);
    byte[] tagBytes = taskBaseName.getBytes();

    TaskOptions options =
        withMethod(TaskOptions.Method.PULL)
            .taskName(taskBaseName + "_0")
            .tag(tagBytes)
            .payload("");
    queue.add(options);
    sync(5000);  // Give tasks a chance to become available.

    List<TaskHandle> tasks = queue.leaseTasksByTagBytes(1, TimeUnit.SECONDS, 10, tagBytes);
    assertEquals(1, tasks.size());
    queue.deleteTask(tasks);
}
 

import com.google.appengine.api.taskqueue.TaskHandle; //导入依赖的package包/类
@Test
public void testPullMultipleWithSameTag() throws Exception {
    String groupTag = "testLeaseTasksByTagBytes";
    String taskBaseName = groupTag + "_" + getTimeStampRandom();
    TaskHandle th1 = queue.add(withMethod(PULL).tag(taskBaseName).payload("foobar", "UTF-8"));
    TaskHandle th2 = queue.add(withMethod(PULL).tag(taskBaseName).payload("foofoo".getBytes(), "UTF-8"));
    sync();
    try {
        int numTasksToLease = 100;
        List<TaskHandle> handles = queue.leaseTasksByTag(30, TimeUnit.MINUTES, numTasksToLease, taskBaseName);
        assertEquals(2, handles.size());

        Set<String> createdHandles = new HashSet<>();
        createdHandles.add(th1.getName());
        createdHandles.add(th2.getName());

        Set<String> returnedHandles = new HashSet<>();
        returnedHandles.add(handles.get(0).getName());
        returnedHandles.add(handles.get(1).getName());

        assertEquals(createdHandles, returnedHandles);
    } finally {
        queue.deleteTask(th1);
        queue.deleteTask(th2);
    }
}
 

import com.google.appengine.api.taskqueue.TaskHandle; //导入依赖的package包/类
@Test
public void testPullMultipleWithDiffTag() throws Exception {
    String groupTag = "testPullMultipleWithDiffTag";
    String taskBaseName = groupTag + "_" + getTimeStampRandom();

    TaskHandle th1 = queue.add(withMethod(PULL).tag(taskBaseName).payload("foobar"));
    TaskHandle th2 = queue.add(withMethod(PULL).tag(taskBaseName + "other").payload("foofoo".getBytes()));
    TaskHandle th3 = queue.add(withMethod(PULL).tag(taskBaseName).payload("foofoo".getBytes()));
    sync();
    try {
        List<TaskHandle> handles = queue.leaseTasksByTag(30, TimeUnit.MINUTES, 100, taskBaseName);
        assertEquals(2, handles.size());

        handles = queue.leaseTasksByTag(30, TimeUnit.MINUTES, 100, taskBaseName + "other");
        assertEquals(1, handles.size());
        assertEquals(th2.getName(), handles.get(0).getName());
    } finally {
        queue.deleteTask(th1);
        queue.deleteTask(th2);
        queue.deleteTask(th3);
    }
}
 

import com.google.appengine.api.taskqueue.TaskHandle; //导入依赖的package包/类
@Test
public void testEtaMillis() {
    String tag = "testEtaMillis_" + getTimeStampRandom();
    queue.add(withMethod(PULL).etaMillis(System.currentTimeMillis() + 10000).tag(tag));
    sync(5000);  // Give tasks a chance to become available.

    List<TaskHandle> tasks = queue.leaseTasks(LeaseOptions.Builder.withTag(tag).leasePeriod(1, TimeUnit.SECONDS).countLimit(1));
    assertEquals(0, tasks.size());

    sync(10000);

    tasks = queue.leaseTasks(LeaseOptions.Builder.withTag(tag.getBytes()).leasePeriod(1, TimeUnit.SECONDS).countLimit(1));
    assertEquals(1, tasks.size());

    queue.deleteTask(tasks);
}
 

import com.google.appengine.api.taskqueue.TaskHandle; //导入依赖的package包/类
@Test
public void testCountdownMillis() {
    String tag = "testCountdownMillis_" + getTimeStampRandom();
    queue.add(withMethod(PULL).countdownMillis(10000).tag(tag));
    sync(5000);  // Give tasks a chance to become available.

    List<TaskHandle> tasks = queue.leaseTasks(LeaseOptions.Builder.withTag(tag.getBytes()).leasePeriod(1, TimeUnit.SECONDS).countLimit(1));
    assertEquals(0, tasks.size());

    sync(10000);

    tasks = queue.leaseTasks(LeaseOptions.Builder.withTag(tag).leasePeriod(1, TimeUnit.SECONDS).countLimit(1));
    assertEquals(1, tasks.size());

    queue.deleteTask(tasks);
}
 

import com.google.appengine.api.taskqueue.TaskHandle; //导入依赖的package包/类
/**
 * @param count        Number of tasks to create.
 * @param taskBaseName add count to this name to create task name.
 * @param groupTag     will be used during tearDown to delete tasks.
 * @param payload      to be used by task.
 * @return List of TaskHandles from queue.add()
 */
private List<TaskHandle> addTasks(int count, String taskBaseName,
                                  String groupTag, String payload) {
    ArrayList<TaskOptions> optionList = new ArrayList<>();
    for (int i = 0; i < count; i++) {
        TaskOptions options =
            withMethod(TaskOptions.Method.PULL)
                .taskName(taskBaseName + "_" + i)
                .tag(groupTag)
                .payload(payload);
        optionList.add(options);
    }
    taskTags.add(groupTag);
    List<TaskHandle> taskHandles = queue.add(optionList);
    sync(5000);  // Give tasks a chance to become available.

    return taskHandles;
}
 

import com.google.appengine.api.taskqueue.TaskHandle; //导入依赖的package包/类
@Test
public void testTaskHandleContainsAllNecessaryProperties() throws Exception {
    String name = "testTaskHandleContainsAllNecessaryProperties-" + System.currentTimeMillis();
    Queue queue = QueueFactory.getDefaultQueue();

    TaskOptions options = withTaskName(name).payload("payload");
    options.etaMillis(0); // TODO -- remove this once NPE is fixewd

    TaskHandle handle = waitOnFuture(queue.addAsync(options));

    assertEquals("default", handle.getQueueName());
    assertEquals(name, handle.getName());
    assertEquals("payload", new String(handle.getPayload(), "UTF-8"));
    assertNotNull(handle.getEtaMillis());
    assertEquals(0, (int) handle.getRetryCount());
}