// TaskWait returns a channel that is closed only when the specified task is
// completed.
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
}
683122
func (p *grpcPeer) monitorLoopWait(ctx context.Context, cancel context.CancelFunc, connectivityState connectivity.State) (connectivity.State, bool) {
	changedC := make(chan bool, 1)
	go func() { changedC <- p.clientConn.WaitForStateChange(ctx, connectivityState) }()

	loop := false
	select {
	case changed := <-changedC:
		if cancel != nil {
			cancel()
		}
		if changed {
			connectivityState = p.clientConn.GetState()
		}
		loop = true
	case <-p.stoppingC:
	case <-p.t.once.Stopping():
		if cancel != nil {
			cancel()
		}
	}
	return connectivityState, loop
}
65500
func (executor *UnboundedExecutor) StopAndWait(ctx context.Context) {
	executor.cancel()
	for {
		oneHundredMilliseconds := time.NewTimer(time.Millisecond * 100)
		select {
		case <-oneHundredMilliseconds.C:
			if executor.checkNoActiveGoroutines() {
				return
			}
		case <-ctx.Done():
			return
		}
	}
}
1081352
func WithWait(d time.Duration) Option {
	return func(l *Loader) {
		l.wait = d
	}
}
479372
func (u *UpdateJob) WaitForBatchCompletion(batchWait bool) *UpdateJob {
	u.req.Settings.WaitForBatchCompletion = batchWait
	return u
}
999516
func (t *Task) Delay(c *Client, d time.Duration) error {
	return c.Delay(t, d)
}
1639
func ReadWait(readWait time.Duration) func(*wsConnection) {
	return func(wsc *wsConnection) {
		wsc.readWait = readWait
	}
}
888702
func (c *Client) Delay(t *Task, delay time.Duration) error {

	client, err := c.pool.Get()

	if err != nil {
		return err
	}
	defer client.Close()

	b, err := t.marshal()
	if err != nil {
		return fmt.Errorf("Could not marshal task: %s", err)
	}

	ar := disque.AddRequest{
		Job: disque.Job{
			Queue: qname(t.Name),
			Data:  b,
		},
		Timeout:   c.enqueueTimeout,
		Replicate: c.replicationFactor,
		Delay:     delay,
		TTL:       t.ttl,
		Retry:     t.retry,
	}

	t.jobId, err = client.Add(ar)

	return err
}
1245949
func (c *Client) TerminateTask(guid string) error {
	req := c.NewRequest("PUT", fmt.Sprintf("/v3/tasks/%s/cancel", guid))
	resp, err := c.DoRequest(req)
	if err != nil {
		return errors.Wrap(err, "Error terminating task")
	}
	defer resp.Body.Close()

	if resp.StatusCode != 202 {
		return errors.Wrapf(err, "Failed terminating task, response status code %d", resp.StatusCode)
	}
	return nil
}
753197
func (s *blockDiskStore) startOpOrWait(id kbfsblock.ID) (
	closeCh chan<- struct{}, waitCh <-chan struct{}) {
	s.lock.Lock()
	defer s.lock.Unlock()

	waitCh = s.puts[id]
	if waitCh == nil {
		// If this caller is getting exclusive access to this `id`,
		// make a channel, store it as receive-only in `puts`, and
		// return it as send-only to the caller, to ensure that only
		// this caller can finish the exclusive access.
		ch := make(chan struct{})
		s.puts[id] = ch
		closeCh = ch
	}

	return closeCh, waitCh
}
405035
func (s *WorkflowExecutionContinuedAsNewEventAttributes) SetTaskStartToCloseTimeout(v string) *WorkflowExecutionContinuedAsNewEventAttributes {
	s.TaskStartToCloseTimeout = &v
	return s
}
253635
func (c *Client) AsyncStart() error {
	if err := c.Listen(); err != nil {
		return err
	}
	var started sync.WaitGroup
	started.Add(1)
	go func() {
		started.Done()
		c.Serve()
	}()
	started.Wait()
	return nil
}
1076055
func (s *WorkflowExecutionStartedEventAttributes) SetTaskStartToCloseTimeout(v string) *WorkflowExecutionStartedEventAttributes {
	s.TaskStartToCloseTimeout = &v
	return s
}
282710
func (o *WaitOptions) RunWait() error {
	visitCount := 0
	err := o.ResourceFinder.Do().Visit(func(info *resource.Info, err error) error {
		if err != nil {
			return err
		}

		visitCount++
		finalObject, success, err := o.ConditionFn(info, o)
		if success {
			o.Printer.PrintObj(finalObject, o.Out)
			return nil
		}
		if err == nil {
			return fmt.Errorf("%v unsatisified for unknown reason", finalObject)
		}
		return err
	})
	if err != nil {
		return err
	}
	if visitCount == 0 {
		return errNoMatchingResources
	}
	return err
}
1064229
func (c *Client) workOnTask(t task) {
	switch t.Action {
	case subscribeAction:
		event := t.Listener.Event
		// Check to see if it already exists in the subscribed map.
		// If not, then we need to create a new listener list and
		// start listening on the pubsub client. Otherwise just
		// append it.
		if _, exists := c.subscribed[event]; !exists || t.Force {
			c.subscribed[event] = []*Listener{t.Listener}
			switch t.Listener.Type {
			case Channel:
				c.pubsub.Subscribe(event)
			case Pattern:
				c.pubsub.PSubscribe(event)
			}
		} else {
			c.subscribed[event] = append(c.subscribed[event], t.Listener)
		}
	case unsubscribeAction:
		event := t.Listener.Event
		// Look for the listener in the subscribed list and,
		// once found, remove it.
		list := c.subscribed[event]
		for i, l := range list {
			if l == t.Listener {
				c.subscribed[event] = append(list[:i], list[i+1:]...)
				break
			}
		}

		// If the list is now empty, we can unsubscribe on Redis and
		// remove it from our registery
		if len(c.subscribed[event]) == 0 || t.Force {
			switch t.Listener.Type {
			case Channel:
				c.pubsub.Unsubscribe(event)
			case Pattern:
				c.pubsub.PUnsubscribe(event)
			}
			delete(c.subscribed, event)
		}
	case disruptAction:
		c.pubsub.Conn.Close()
	default:
		panic("unknown task")
	}
}
761231
func (i *Index) WaitTask(taskID int) error {
	return waitWithRetry(func() (bool, error) {
		res, err := i.GetStatus(taskID)
		if err != nil {
			return true, err
		}
		if res.Status == "published" {
			return true, nil
		}
		return false, nil
	})
}
834567
func (f *Fibre) SetWait(wait string) {
	f.wait, _ = time.ParseDuration(wait)
}