// Task.Result() returns nil
335797
func (m *Manager) DeleteQueryAsync(
	name string,
	deleteEmittedStreams bool,
	userCredentials *cli.UserCredentials,
) *tasks.Task {
	if name == "" {
		panic("name must be present")
	}

	return m.client.Delete(m.httpEndpoint, name, deleteEmittedStreams, userCredentials)
}
1345644
func (c *Client) GetTask(id int) (Task, error) {
	task := Task{}
	stringID := strconv.Itoa(id)
	r := c.NewRequest("GET", "/tasks/"+stringID)
	resp, err := c.DoRequest(r)

	if err != nil {
		log.Printf("Error requesting task %v", err)
		return task, err
	}
	defer resp.Body.Close()

	resBody, err := ioutil.ReadAll(resp.Body)
	if err != nil {
		log.Printf("Error reading task request %v", resBody)
		return task, err
	}
	err = json.Unmarshal(resBody, &task)
	if err != nil {
		log.Printf("Error unmarshaling task %v", err)
	}
	return task, err
}
1301246
func (b *AMQPCeleryBackend) GetResult(taskID string) (*ResultMessage, error) {

	queueName := strings.Replace(taskID, "-", "", -1)

	args := amqp.Table{"x-expires": int32(86400000)}

	_, err := b.QueueDeclare(
		queueName, // name
		true,      // durable
		true,      // autoDelete
		false,     // exclusive
		false,     // noWait
		args,      // args
	)
	if err != nil {
		return nil, err
	}

	err = b.ExchangeDeclare(
		"default",
		"direct",
		true,
		true,
		false,
		false,
		nil,
	)
	if err != nil {
		return nil, err
	}

	// open channel temporarily
	channel, err := b.Consume(queueName, "", false, false, false, false, nil)
	if err != nil {
		return nil, err
	}

	var resultMessage ResultMessage

	delivery := <-channel
	deliveryAck(delivery)
	if err := json.Unmarshal(delivery.Body, &resultMessage); err != nil {
		return nil, err
	}
	return &resultMessage, nil
}
1079801
func (c *Client) Result(req *Request, resp interface{}) error {
	result, err := c.Do(req)
	if err != nil {
		return err
	}
	return unmarshal(result, resp)
}
667780
func (ctr *ContainerTaskRels) Get(containerID string) *TaskInfo {
	ctr.Lock()
	defer ctr.Unlock()
	return ctr.rels[containerID]
}
1060519
func (f *Future) GetResult() (interface{}, error) {
	f.lock.Lock()
	if f.triggered {
		f.lock.Unlock()
		return f.item, f.err
	}
	f.lock.Unlock()

	f.wg.Wait()
	return f.item, f.err
}
37100
func (c *SearchCall) Result(result string) *SearchCall {
	c.urlParams_.Set("result", result)
	return c
}
1220036
func (d *Dispatcher) Tasks(r *api.TasksRequest, stream api.Dispatcher_TasksServer) error {
	d.rpcRW.RLock()
	defer d.rpcRW.RUnlock()

	dctx, err := d.isRunningLocked()
	if err != nil {
		return err
	}

	nodeInfo, err := ca.RemoteNode(stream.Context())
	if err != nil {
		return err
	}
	nodeID := nodeInfo.NodeID

	fields := logrus.Fields{
		"node.id":      nodeID,
		"node.session": r.SessionID,
		"method":       "(*Dispatcher).Tasks",
	}
	if nodeInfo.ForwardedBy != nil {
		fields["forwarder.id"] = nodeInfo.ForwardedBy.NodeID
	}
	log.G(stream.Context()).WithFields(fields).Debug("")

	if _, err = d.nodes.GetWithSession(nodeID, r.SessionID); err != nil {
		return err
	}

	tasksMap := make(map[string]*api.Task)
	nodeTasks, cancel, err := store.ViewAndWatch(
		d.store,
		func(readTx store.ReadTx) error {
			tasks, err := store.FindTasks(readTx, store.ByNodeID(nodeID))
			if err != nil {
				return err
			}
			for _, t := range tasks {
				tasksMap[t.ID] = t
			}
			return nil
		},
		api.EventCreateTask{Task: &api.Task{NodeID: nodeID},
			Checks: []api.TaskCheckFunc{api.TaskCheckNodeID}},
		api.EventUpdateTask{Task: &api.Task{NodeID: nodeID},
			Checks: []api.TaskCheckFunc{api.TaskCheckNodeID}},
		api.EventDeleteTask{Task: &api.Task{NodeID: nodeID},
			Checks: []api.TaskCheckFunc{api.TaskCheckNodeID}},
	)
	if err != nil {
		return err
	}
	defer cancel()

	for {
		if _, err := d.nodes.GetWithSession(nodeID, r.SessionID); err != nil {
			return err
		}

		var tasks []*api.Task
		for _, t := range tasksMap {
			// dispatcher only sends tasks that have been assigned to a node
			if t != nil && t.Status.State >= api.TaskStateAssigned {
				tasks = append(tasks, t)
			}
		}

		if err := stream.Send(&api.TasksMessage{Tasks: tasks}); err != nil {
			return err
		}

		// bursty events should be processed in batches and sent out snapshot
		var (
			modificationCnt int
			batchingTimer   *time.Timer
			batchingTimeout <-chan time.Time
		)

	batchingLoop:
		for modificationCnt < modificationBatchLimit {
			select {
			case event := <-nodeTasks:
				switch v := event.(type) {
				case api.EventCreateTask:
					tasksMap[v.Task.ID] = v.Task
					modificationCnt++
				case api.EventUpdateTask:
					if oldTask, exists := tasksMap[v.Task.ID]; exists {
						// States ASSIGNED and below are set by the orchestrator/scheduler,
						// not the agent, so tasks in these states need to be sent to the
						// agent even if nothing else has changed.
						if equality.TasksEqualStable(oldTask, v.Task) && v.Task.Status.State > api.TaskStateAssigned {
							// this update should not trigger action at agent
							tasksMap[v.Task.ID] = v.Task
							continue
						}
					}
					tasksMap[v.Task.ID] = v.Task
					modificationCnt++
				case api.EventDeleteTask:
					delete(tasksMap, v.Task.ID)
					modificationCnt++
				}
				if batchingTimer != nil {
					batchingTimer.Reset(batchingWaitTime)
				} else {
					batchingTimer = time.NewTimer(batchingWaitTime)
					batchingTimeout = batchingTimer.C
				}
			case <-batchingTimeout:
				break batchingLoop
			case <-stream.Context().Done():
				return stream.Context().Err()
			case <-dctx.Done():
				return dctx.Err()
			}
		}

		if batchingTimer != nil {
			batchingTimer.Stop()
		}
	}
}
962912
func (s *globalTaskService) TaskWaitC(taskID int) <-chan int {
	c := make(chan int)

	go func() {
		defer close(c)

		s.RLock()
		t, ok := s.tasks[taskID]
		s.RUnlock()

		if !ok {
			return
		}

		// remove the task from the queue after a configured about of time
		defer s.taskRemoveAfter(t)

		// signal that the task is complete
		<-t.done
	}()

	return c
}
662026
func (w *Worker) TaskQueue(tasks ...Task) (taskIds []string, err error) {
	outTasks := make([]map[string]interface{}, 0, len(tasks))

	for _, task := range tasks {
		thisTask := map[string]interface{}{
			"code_name": task.CodeName,
			"payload":   task.Payload,
			"priority":  task.Priority,
			"cluster":   task.Cluster,
			"label":     task.Label,
		}
		if task.Timeout != nil {
			thisTask["timeout"] = (*task.Timeout).Seconds()
		}
		if task.Delay != nil {
			thisTask["delay"] = int64((*task.Delay).Seconds())
		}

		outTasks = append(outTasks, thisTask)
	}

	in := map[string][]map[string]interface{}{"tasks": outTasks}
	out := struct {
		Tasks []struct {
			Id string `json:"id"`
		} `json:"tasks"`
		Msg string `json:"msg"`
	}{}

	err = w.tasks().Req("POST", &in, &out)
	if err != nil {
		return
	}

	taskIds = make([]string, 0, len(out.Tasks))
	for _, task := range out.Tasks {
		taskIds = append(taskIds, task.Id)
	}

	return
}
879700
func (r Result) Get() (runtime.Object, error) {
	if r.err != nil {
		// Check whether the result has a Status object in the body and prefer that.
		return nil, r.Error()
	}
	if r.decoder == nil {
		return nil, fmt.Errorf("serializer for %s doesn't exist", r.contentType)
	}

	// decode, but if the result is Status return that as an error instead.
	out, _, err := r.decoder.Decode(r.body, nil, nil)
	if err != nil {
		return nil, err
	}
	switch t := out.(type) {
	case *metav1.Status:
		// any status besides StatusSuccess is considered an error.
		if t.Status != metav1.StatusSuccess {
			return nil, errors.FromObject(t)
		}
	}
	return out, nil
}
87819
func (s *Scheduler) Tasks(ctx context.Context, app string) ([]*twelvefactor.Task, error) {
	var instances []*twelvefactor.Task

	tasks, err := s.tasks(app)
	if err != nil {
		return nil, err
	}

	taskDefinitions := make(map[string]*ecs.TaskDefinition)
	for _, t := range tasks {
		k := *t.TaskDefinitionArn

		if _, ok := taskDefinitions[k]; !ok {
			resp, err := s.ecs.DescribeTaskDefinition(&ecs.DescribeTaskDefinitionInput{
				TaskDefinition: t.TaskDefinitionArn,
			})
			if err != nil {
				return nil, err
			}
			taskDefinitions[k] = resp.TaskDefinition
		}
	}

	// Map from clusterARN to containerInstanceARN pointers to batch tasks from the same cluster
	clusterMap := make(map[string][]*string)

	for _, t := range tasks {
		k := *t.ClusterArn
		clusterMap[k] = append(clusterMap[k], t.ContainerInstanceArn)
	}

	// Map from containerInstanceARN to ec2-instance-id
	hostMap := make(map[string]string)

	for clusterArn, containerArnPtrs := range clusterMap {
		for _, chunk := range chunkStrings(containerArnPtrs, MaxDescribeContainerInstances) {
			resp, err := s.ecs.DescribeContainerInstances(&ecs.DescribeContainerInstancesInput{
				Cluster:            aws.String(clusterArn),
				ContainerInstances: chunk,
			})
			if err != nil {
				return nil, err
			}
			for _, f := range resp.Failures {
				return nil, fmt.Errorf("error describing container instance %s: %s", aws.StringValue(f.Arn), aws.StringValue(f.Reason))
			}

			for _, ci := range resp.ContainerInstances {
				hostMap[aws.StringValue(ci.ContainerInstanceArn)] = aws.StringValue(ci.Ec2InstanceId)
			}
		}
	}

	for _, t := range tasks {
		taskDefinition := taskDefinitions[*t.TaskDefinitionArn]

		id, err := arn.ResourceID(*t.TaskArn)
		if err != nil {
			return instances, err
		}

		hostId := hostMap[*t.ContainerInstanceArn]

		p, err := taskDefinitionToProcess(taskDefinition)
		if err != nil {
			return instances, err
		}

		state := aws.StringValue(t.LastStatus)
		var updatedAt time.Time
		switch state {
		case "PENDING":
			updatedAt = *t.CreatedAt
		case "RUNNING":
			updatedAt = *t.StartedAt
		case "STOPPED":
			updatedAt = *t.StoppedAt
		}

		instances = append(instances, &twelvefactor.Task{
			Process:   p,
			State:     state,
			ID:        id,
			Host:      twelvefactor.Host{ID: hostId},
			UpdatedAt: updatedAt,
		})
	}

	return instances, nil
}
962647
func (a AsyncJobResult) Result(i interface{}) error {
	if a.JobStatus == Failure {
		return a.Error()
	}

	if a.JobStatus == Success {
		m := map[string]json.RawMessage{}
		err := json.Unmarshal(*(a.JobResult), &m)

		if err == nil {
			if len(m) >= 1 {
				if _, ok := m["success"]; ok {
					return json.Unmarshal(*(a.JobResult), i)
				}

				// otherwise, pick the first key
				for k := range m {
					return json.Unmarshal(m[k], i)
				}
			}
			return errors.New("empty response")
		}
	}

	return nil
}
490065
func Task(t *api.Task) string {
	if t.Annotations.Name != "" {
		// if set, use the container Annotations.Name field, set in the orchestrator.
		return t.Annotations.Name
	}

	slot := fmt.Sprint(t.Slot)
	if slot == "" || t.Slot == 0 {
		// when no slot id is assigned, we assume that this is node-bound task.
		slot = t.NodeID
	}

	// fallback to service.instance.id.
	return fmt.Sprintf("%s.%s.%s", t.ServiceAnnotations.Name, slot, t.ID)
}
560379
func Run(jobs []Tasker) (err error) {
	// []T does not convert to []Tasker implicitly even is T implements
	// Tasker. We need to iterate on []Tasker making an explicit cast.
	// http://golang.org/doc/faq#convert_slice_of_interface
	prematureEnd := make(chan struct{})
	jobsQueue := make(chan Tasker, workersNumber)
	done := make(chan struct{}, workersNumber)
	var totalDone int
	go populateQueue(jobsQueue, jobs, prematureEnd)
	go parallelizeWorkers(jobsQueue, done)
	// TODO add a case timeout that returns error.
	for {
		select {
		case <-done:
			totalDone++
		case <-prematureEnd:
			err = ErrTasksNotCompleted
		}
		if totalDone == workersNumber {
			// We can assume that jobsQueue is closed and
			// that no goroutine is operating on []Tasker.
			break
		}
	}
	return
}
1348097
func (rb *ResultBuilder) Results() Results {
	return &results{
		query: rb.Query,
		proc:  rb.Process,
		res:   rb.Output,
	}
}
175156
func (e TaskError) Error() string {
	return fmt.Sprintf("photon: Task '%s' is in error state: {@step==%s}", e.ID, GetStep(e.Step))
}