Method consumer

task/list.go:43–100

consumer is processing the queue

()

// consumer is processing the queue
func (list *List) consumer() {
	for {
		select {
		case task := <-list.queue:
			// Set task state to RUNNING before processing
			list.Lock()
			task.State = RUNNING
			list.Unlock()

			go func() {
				retValue, err := task.process(aptly.Progress(task.output), task.detail)

				// Update task completion state and cleanup with list lock held
				list.Lock()
				{
					if err != nil {
						task.output.Printf("Task failed with error: %v", err)
						task.State = FAILED
						task.err = err
						task.processReturnValue = retValue
					} else {
						task.output.Print("Task succeeded")
						task.State = SUCCEEDED
						task.err = nil
						task.processReturnValue = retValue
					}

					list.usedResources.Free(task.Resources)

					task.wgTask.Done()
					list.wg.Done()

					unlocked := false
					for _, t := range list.tasks {
						if t.State == IDLE {
							// check resources
							blockingTasks := list.usedResources.UsedBy(t.Resources)
							if len(blockingTasks) == 0 {
								list.usedResources.MarkInUse(t.Resources, t)
								// unlock list since queueing may block
								list.Unlock()
								unlocked = true
								list.queue <- t
								break
							}
						}
					}
					if !unlocked {
						list.Unlock()
					}
				}
			}()

		case <-list.queueDone:
			return
		}
	}
}