// Respond to a watch with the snapshot value. The value channel should have capacity not to block.
// TODO(kuat) do not respond always, see issue https://github.com/envoyproxy/go-control-plane/issues/46
559648
func (cache *snapshotCache) respond(request Request, value chan Response, resources map[string]Resource, version string) {
	// for ADS, the request names must match the snapshot names
	// if they do not, then the watch is never responded, and it is expected that envoy makes another request
	if len(request.ResourceNames) != 0 && cache.ads {
		if err := superset(nameSet(request.ResourceNames), resources); err != nil {
			if cache.log != nil {
				cache.log.Infof("ADS mode: not responding to request: %v", err)
			}
			return
		}
	}
	if cache.log != nil {
		cache.log.Infof("respond %s%v version %q with version %q",
			request.TypeUrl, request.ResourceNames, request.VersionInfo, version)
	}

	value <- createResponse(request, resources, version)
}
545236
func (c *APIClient) WatchServices(respCh chan SvcWatchResp) {
	ctx, _ := context.WithCancel(context.Background())

	go func() {
		// Make request to Kubernetes API
		getURL := c.watchBase + "services"
		req, err := http.NewRequest("GET", getURL, nil)
		if err != nil {
			respCh <- SvcWatchResp{opcode: "FATAL", errStr: fmt.Sprintf("Req %v", err)}
			return
		}
		if len(strings.TrimSpace(c.authToken)) > 0 {
			req.Header.Set("Authorization", fmt.Sprintf("Bearer %s", c.authToken))
		}
		res, err := ctxhttp.Do(ctx, c.client, req)
		if err != nil {
			log.Errorf("Watch error: %v", err)
			respCh <- SvcWatchResp{opcode: "FATAL", errStr: fmt.Sprintf("Do %v", err)}
			return
		}
		defer res.Body.Close()

		var wss watchSvcStatus
		reader := bufio.NewReader(res.Body)

		// bufio.Reader.ReadBytes is blocking, so we watch for
		// context timeout or cancellation in a goroutine
		// and close the response body when see see it. The
		// response body is also closed via defer when the
		// request is made, but closing twice is OK.
		go func() {
			<-ctx.Done()
			res.Body.Close()
		}()

		for {
			line, err := reader.ReadBytes('\n')
			if ctx.Err() != nil {
				respCh <- SvcWatchResp{opcode: "ERROR", errStr: fmt.Sprintf("ctx %v", err)}
				return
			}
			if err != nil {
				respCh <- SvcWatchResp{opcode: "ERROR", errStr: fmt.Sprintf("read %v", err)}
				return
			}
			if err := json.Unmarshal(line, &wss); err != nil {
				respCh <- SvcWatchResp{opcode: "WARN", errStr: fmt.Sprintf("unmarshal %v", err)}
				continue
			}

			if wss.Object.ObjectMeta.Namespace == "kube-system" && (wss.Object.ObjectMeta.Name == "kube-scheduler" || wss.Object.ObjectMeta.Name == "kube-controller-manager") {
				// Ignoring these frequent updates
				continue
			}

			resp := SvcWatchResp{opcode: wss.Type}
			resp.svcName = wss.Object.ObjectMeta.Name
			sSpec := core.ServiceSpec{}
			sSpec.Ports = make([]core.PortSpec, 0, 1)
			sSpec.IPAddress = wss.Object.Spec.ClusterIP
			sSpec.ExternalIPs = wss.Object.Spec.ExternalIPs
			for _, port := range wss.Object.Spec.Ports {
				ps := core.PortSpec{Protocol: string(port.Protocol),
					SvcPort:  uint16(port.Port),
					NodePort: uint16(port.NodePort),
				}

				// handle 'kubernetes' service
				// Use port from configuration till named ports are supported.
				if resp.svcName == "kubernetes" && (len(wss.Object.ObjectMeta.Namespace) == 0 ||
					wss.Object.ObjectMeta.Namespace == "default") {
					ps.ProvPort = c.apiServerPort
				} else {
					ps.ProvPort = uint16(port.TargetPort)
				}
				sSpec.Ports = append(sSpec.Ports, ps)
			}

			resp.svcSpec = sSpec
			log.Infof("resp: %+v", resp)

			respCh <- resp
		}
	}()
}
1112240
func (c *Client) sendToChannel(keyed keyedData) {
	c.Lock()
	defer c.Unlock()

	for key, channel := range c.watchers {
		if strings.Contains(keyed.Key, key) {
			channel <- keyed
			return
		}
	}
}
1126741
func (w *HubWatcher) Watch(collection string, id interface{}, ch chan<- Change) {
	if id == nil {
		panic("watcher: cannot watch a document with nil id")
	}
	// We use a value of -2 to indicate that we don't know the state of the document.
	// -1 would indicate that we think the document is deleted (and won't trigger
	// a change event if the document really is deleted).
	w.sendAndWaitReq(reqWatch{
		key:          watchKey{collection, id},
		info:         watchInfo{ch, -2, nil},
		registeredCh: make(chan error),
	})
}
246813
func (w *Watcher) flush() {
	// w.pending may get new requests as we handle other requests.
	for i := 0; i < len(w.pending); i++ {
		e := &w.pending[i]
		// Allow the handling of requests while waiting for e.ch
		// to be ready to read from the channel.
		for e.ch != nil {
			select {
			case <-w.tomb.Dying():
				return
			case req := <-w.request:
				w.handle(req)
				// handle may append to the w.pending array, and/or it may unwatch something that was previously pending
				// thus changing e.ch to nil while we are waiting to send the request. We need to make sure we are using
				// the correct 'e' object
				// See TestRobustness which fails if this line doesn't exist
				e = &w.pending[i]
				continue
			case e.ch <- Change{e.key, e.alive}:
			}
			break
		}
	}
	w.pending = w.pending[:0]
}
438569
func (p *Project) Watch(wg *sync.WaitGroup) {
	var err error
	// change channel
	p.stop = make(chan bool)
	// init a new watcher
	p.watcher, err = NewFileWatcher(p.parent.Settings.Legacy)
	if err != nil {
		log.Fatal(err)
	}
	defer func() {
		close(p.stop)
		p.watcher.Close()
	}()
	// before start checks
	p.Before()
	// start watcher
	go p.Reload("", p.stop)
L:
	for {
		select {
		case event := <-p.watcher.Events():
			if p.parent.Settings.Recovery.Events {
				log.Println("File:", event.Name, "LastFile:", p.last.file, "Time:", time.Now(), "LastTime:", p.last.time)
			}
			if time.Now().Truncate(time.Second).After(p.last.time) {
				// switch event type
				switch event.Op {
				case fsnotify.Chmod:
				case fsnotify.Remove:
					p.watcher.Remove(event.Name)
					if p.Validate(event.Name, false) && ext(event.Name) != "" {
						// stop and restart
						close(p.stop)
						p.stop = make(chan bool)
						p.Change(event)
						go p.Reload("", p.stop)
					}
				default:
					if p.Validate(event.Name, true) {
						fi, err := os.Stat(event.Name)
						if err != nil {
							continue
						}
						if fi.IsDir() {
							filepath.Walk(event.Name, p.walk)
						} else {
							// stop and restart
							close(p.stop)
							p.stop = make(chan bool)
							p.Change(event)
							go p.Reload(event.Name, p.stop)
							p.last.time = time.Now().Truncate(time.Second)
							p.last.file = event.Name
						}
					}
				}
			}
		case err := <-p.watcher.Errors():
			p.Err(err)
		case <-p.exit:
			p.After()
			break L
		}
	}
	wg.Done()
}
4125
func (r *InitRequest) Respond(resp *InitResponse) {
	out := &initOut{
		outHeader:    outHeader{Unique: uint64(r.ID)},
		Major:        kernelVersion,
		Minor:        kernelMinorVersion,
		MaxReadahead: resp.MaxReadahead,
		Flags:        uint32(resp.Flags),
		MaxWrite:     resp.MaxWrite,
	}
	r.Conn.respond(&out.outHeader, unsafe.Sizeof(*out))
}
256017
func (c *Client) Watch(resp func(response keyval.BytesWatchResp), closeChan chan string, keys ...string) error {
	c.Lock()
	defer c.Unlock()

	for _, key := range keys {
		dc := make(chan keyedData)
		go c.watch(resp, dc, closeChan, key)
		c.watchers[key] = dc
	}

	return nil
}
265014
func (s *Stream) Listen() <-chan Event {
	ch := make(chan Event)
	go func(s *Stream, ch chan Event) {
		for {
			// sleep for a while before the new polling tick,
			// otherwise operation is too intensive and blocking
			time.Sleep(10 * time.Millisecond)
			events, err := s.Read(1024)
			// Note: It's not very reasonable to push sliced data into
			// a channel, several perf penalities there are.
			// This function is added as a handy utility.
			if err != nil {
				continue
			}
			for i := range events {
				ch <- events[i]
			}
		}
	}(s, ch)
	return ch
}
165695
func (r *SetattrRequest) Respond(resp *SetattrResponse) {
	out := &attrOut{
		outHeader:     outHeader{Unique: uint64(r.ID)},
		AttrValid:     uint64(resp.AttrValid / time.Second),
		AttrValidNsec: uint32(resp.AttrValid % time.Second / time.Nanosecond),
		Attr:          resp.Attr.attr(),
	}
	r.Conn.respond(&out.outHeader, unsafe.Sizeof(*out))
}
1199682
func (r *ReadRequest) Respond(resp *ReadResponse) {
	out := &outHeader{Unique: uint64(r.ID)}
	r.Conn.respondData(out, unsafe.Sizeof(*out), resp.Data)
}
1022807
func (s *SQS) Listen(fn dispatch.Listenner) error {

	errChan := make(chan error)

	go func() {

		for true {
			params := &sqs.ReceiveMessageInput{
				QueueUrl:            s.queueURL,
				MaxNumberOfMessages: aws.Int64(1),
			}

			if resp, err := s.client().ReceiveMessage(params); err != nil {
				errChan <- err
				break

			} else {
				for _, m := range resp.Messages {

					if m.Body != nil {
						if decoded, err := base64.StdEncoding.DecodeString(*m.Body); err != nil {
							errChan <- err
							return

						} else if err := fn(decoded); err != nil {
							errChan <- err
							return

						} else if err := s.ack(m); err != nil {
							errChan <- err
							return

						}
					}
				}
			}
		}
	}()

	return <-errChan

}
1034864
func (b *BatchFuture) Respond(index uint64, err error) {
	b.index = index
	b.err = err
	close(b.doneCh)
}
1115996
func (rw *RetryWatcher) doReceive() (bool, time.Duration) {
	watcher, err := rw.watcherClient.Watch(metav1.ListOptions{
		ResourceVersion: rw.lastResourceVersion,
	})
	// We are very unlikely to hit EOF here since we are just establishing the call,
	// but it may happen that the apiserver is just shutting down (e.g. being restarted)
	// This is consistent with how it is handled for informers
	switch err {
	case nil:
		break

	case io.EOF:
		// watch closed normally
		return false, 0

	case io.ErrUnexpectedEOF:
		klog.V(1).Infof("Watch closed with unexpected EOF: %v", err)
		return false, 0

	default:
		msg := "Watch failed: %v"
		if net.IsProbableEOF(err) {
			klog.V(5).Infof(msg, err)
			// Retry
			return false, 0
		}

		klog.Errorf(msg, err)
		// Retry
		return false, 0
	}

	if watcher == nil {
		klog.Error("Watch returned nil watcher")
		// Retry
		return false, 0
	}

	ch := watcher.ResultChan()
	defer watcher.Stop()

	for {
		select {
		case <-rw.stopChan:
			klog.V(4).Info("Stopping RetryWatcher.")
			return true, 0
		case event, ok := <-ch:
			if !ok {
				klog.V(4).Infof("Failed to get event! Re-creating the watcher. Last RV: %s", rw.lastResourceVersion)
				return false, 0
			}

			// We need to inspect the event and get ResourceVersion out of it
			switch event.Type {
			case watch.Added, watch.Modified, watch.Deleted, watch.Bookmark:
				metaObject, ok := event.Object.(resourceVersionGetter)
				if !ok {
					_ = rw.send(watch.Event{
						Type:   watch.Error,
						Object: &apierrors.NewInternalError(errors.New("retryWatcher: doesn't support resourceVersion")).ErrStatus,
					})
					// We have to abort here because this might cause lastResourceVersion inconsistency by skipping a potential RV with valid data!
					return true, 0
				}

				resourceVersion := metaObject.GetResourceVersion()
				if resourceVersion == "" {
					_ = rw.send(watch.Event{
						Type:   watch.Error,
						Object: &apierrors.NewInternalError(fmt.Errorf("retryWatcher: object %#v doesn't support resourceVersion", event.Object)).ErrStatus,
					})
					// We have to abort here because this might cause lastResourceVersion inconsistency by skipping a potential RV with valid data!
					return true, 0
				}

				// All is fine; send the event and update lastResourceVersion
				ok = rw.send(event)
				if !ok {
					return true, 0
				}
				rw.lastResourceVersion = resourceVersion

				continue

			case watch.Error:
				// This round trip allows us to handle unstructured status
				errObject := apierrors.FromObject(event.Object)
				statusErr, ok := errObject.(*apierrors.StatusError)
				if !ok {
					klog.Error(spew.Sprintf("Received an error which is not *metav1.Status but %#+v", event.Object))
					// Retry unknown errors
					return false, 0
				}

				status := statusErr.ErrStatus

				statusDelay := time.Duration(0)
				if status.Details != nil {
					statusDelay = time.Duration(status.Details.RetryAfterSeconds) * time.Second
				}

				switch status.Code {
				case http.StatusGone:
					// Never retry RV too old errors
					_ = rw.send(event)
					return true, 0

				case http.StatusGatewayTimeout, http.StatusInternalServerError:
					// Retry
					return false, statusDelay

				default:
					// We retry by default. RetryWatcher is meant to proceed unless it is certain
					// that it can't. If we are not certain, we proceed with retry and leave it
					// up to the user to timeout if needed.

					// Log here so we have a record of hitting the unexpected error
					// and we can whitelist some error codes if we missed any that are expected.
					klog.V(5).Info(spew.Sprintf("Retrying after unexpected error: %#+v", event.Object))

					// Retry
					return false, statusDelay
				}

			default:
				klog.Errorf("Failed to recognize Event type %q", event.Type)
				_ = rw.send(watch.Event{
					Type:   watch.Error,
					Object: &apierrors.NewInternalError(fmt.Errorf("retryWatcher failed to recognize Event type %q", event.Type)).ErrStatus,
				})
				// We are unable to restart the watch and have to stop the loop or this might cause lastResourceVersion inconsistency by skipping a potential RV with valid data!
				return true, 0
			}
		}
	}
}
364120
func (s *subscription) Listen() (interface{}, error) {
	select {
	case <-time.After(s.ttl):
		return nil, errors.New("timed out")
	case item := <-s.c:
		return item, nil
	}
}
146560
func (cw *CloudWatch) Send() error {
	cw.mtx.RLock()
	defer cw.mtx.RUnlock()
	now := time.Now()

	var datums []cloudwatch.MetricDatum

	cw.counters.Reset().Walk(func(name string, lvs lv.LabelValues, values []float64) bool {
		datums = append(datums, cloudwatch.MetricDatum{
			MetricName:      aws.String(name),
			Dimensions:      makeDimensions(lvs...),
			StatisticValues: stats(values),
			Timestamp:       aws.Time(now),
		})
		return true
	})

	var batches [][]cloudwatch.MetricDatum
	for len(datums) > 0 {
		var batch []cloudwatch.MetricDatum
		lim := len(datums)
		if lim > maxConcurrentRequests {
			lim = maxConcurrentRequests
		}
		batch, datums = datums[:lim], datums[lim:]
		batches = append(batches, batch)
	}

	var g errgroup.Group
	for _, batch := range batches {
		batch := batch
		g.Go(func() error {
			cw.sem <- struct{}{}
			defer func() {
				<-cw.sem
			}()
			req := cw.svc.PutMetricDataRequest(&cloudwatch.PutMetricDataInput{
				Namespace:  aws.String(cw.namespace),
				MetricData: batch,
			})
			_, err := req.Send(context.TODO())
			return err
		})
	}
	return g.Wait()
}
567829
func (r *OpenRequest) Respond(resp *OpenResponse) {
	buf := newBuffer(unsafe.Sizeof(openOut{}))
	out := (*openOut)(buf.alloc(unsafe.Sizeof(openOut{})))
	out.Fh = uint64(resp.Handle)
	out.OpenFlags = uint32(resp.Flags)
	r.respond(buf)
}