Function newAddReq

v2/taskqueue/taskqueue.go:202–274

(c context.Context, task *Task, queueName string)

}

func newAddReq(c context.Context, task *Task, queueName string) (*pb.TaskQueueAddRequest, error) {
	if queueName == "" {
		queueName = "default"
	}
	path := task.Path
	if path == "" {
		path = "/_ah/queue/" + queueName
	}
	eta := task.ETA
	if eta.IsZero() {
		eta = time.Now().Add(task.Delay)
	} else if task.Delay != 0 {
		panic("taskqueue: both Delay and ETA are set")
	}
	req := &pb.TaskQueueAddRequest{
		QueueName: []byte(queueName),
		TaskName:  []byte(task.Name),
		EtaUsec:   proto.Int64(eta.UnixNano() / 1e3),
	}
	method := task.method()
	if method == "PULL" {
		// Pull-based task
		req.Body = task.Payload
		req.Mode = pb.TaskQueueMode_PULL.Enum()
		if task.Tag != "" {
			req.Tag = []byte(task.Tag)
		}
	} else {
		// HTTP-based task
		if v, ok := pb.TaskQueueAddRequest_RequestMethod_value[method]; ok {
			req.Method = pb.TaskQueueAddRequest_RequestMethod(v).Enum()
		} else {
			return nil, fmt.Errorf("taskqueue: bad method %q", method)
		}
		req.Url = []byte(path)
		for k, vs := range task.Header {
			for _, v := range vs {
				req.Header = append(req.Header, &pb.TaskQueueAddRequest_Header{
					Key:   []byte(k),
					Value: []byte(v),
				})
			}
		}
		if method == "POST" || method == "PUT" {
			req.Body = task.Payload
		}

		// Namespace headers.
		if _, ok := task.Header[currentNamespace]; !ok {
			// Fetch the current namespace of this request.
			ns := internal.NamespaceFromContext(c)
			req.Header = append(req.Header, &pb.TaskQueueAddRequest_Header{
				Key:   []byte(currentNamespace),
				Value: []byte(ns),
			})
		}
		if _, ok := task.Header[defaultNamespace]; !ok {
			// Fetch the X-AppEngine-Default-Namespace header of this request.