// Stop TODO(rjeczalik)
//
// TODO(rjeczalik): Split parent watchpoint - transfer watches to children
// if parent is no longer needed. This carries a risk that underlying
// watcher calls could fail - reconsider if it's worth the effort.
554650
func (t *recursiveTree) Stop(c chan<- EventInfo) {
	var err error
	fn := func(nd node) (e error) {
		diff := watchDel(nd, c, all)
		switch {
		case diff == none && watchTotal(nd) == 0:
			// TODO(rjeczalik): There's no watchpoints deeper in the tree,
			// probably we should remove the nodes as well.
			return nil
		case diff == none:
			// Removing c from nd does not require shrinking its eventset.
		case diff[1] == 0:
			if watchIsRecursive(nd) {
				e = t.w.RecursiveUnwatch(nd.Name)
			} else {
				e = t.w.Unwatch(nd.Name)
			}
		default:
			if watchIsRecursive(nd) {
				e = t.w.RecursiveRewatch(nd.Name, nd.Name, diff[0], diff[1])
			} else {
				e = t.w.Rewatch(nd.Name, diff[0], diff[1])
			}
		}
		fn := func(nd node) error {
			watchDel(nd, c, all)
			return nil
		}
		err = nonil(err, e, nd.Walk(fn))
		// TODO(rjeczalik): if e != nil store dummy chan in nd.Watch just to
		// retry un/rewatching next time and/or let the user handle the failure
		// vie Error event?
		return errSkip
	}
	t.rw.Lock()
	e := t.root.Walk("", fn) // TODO(rjeczalik): use max root per c
	t.rw.Unlock()
	if e != nil {
		err = nonil(err, e)
	}
	dbgprintf("Stop(%p) error: %v\n", c, err)
}
296150
func (u *UpdateChecker) Stop() {
	if u.ticker != nil {
		u.ticker.Stop()
		u.ticker = nil
	}
}
184580
func (w *Watcher) Close() {
	w.cmdChan <- watcherCmd{
		pin:    Pin{},
		action: watcherClose,
	}
}
917247
func (w *Watcher) Close() error {
	w.exit <- true
	w.consume()
	w.conn.noIdle()

	<-w.done // wait for idle to finish and channels to close
	// At this point, watch goroutine has ended,
	// so it's safe to close connection.
	return w.conn.Close()
}
345761
func (w *commonWatcher) commonLoop() {
	defer close(w.in)
	var wg sync.WaitGroup
	wg.Add(1)
	go func() {
		// When the watcher has been stopped, we send a Stop request
		// to the server, which will remove the watcher and return a
		// CodeStopped error to any currently outstanding call to
		// Next. If a call to Next happens just after the watcher has
		// been stopped, we'll get a CodeNotFound error; Either way
		// we'll return, wait for the stop request to complete, and
		// the watcher will die with all resources cleaned up.
		defer wg.Done()
		<-w.tomb.Dying()
		if err := w.call("Stop", nil); err != nil {
			// Don't log an error if a watcher is stopped due to an agent restart.
			if err.Error() != worker.ErrRestartAgent.Error() && err.Error() != rpc.ErrShutdown.Error() {
				logger.Errorf("error trying to stop watcher: %v", err)
			}
		}
	}()
	wg.Add(1)
	go func() {
		// Because Next blocks until there are changes, we need to
		// call it in a separate goroutine, so the watcher can be
		// stopped normally.
		defer wg.Done()
		for {
			result := w.newResult()
			err := w.call("Next", &result)
			if err != nil {
				if params.IsCodeStopped(err) || params.IsCodeNotFound(err) {
					if w.tomb.Err() != tomb.ErrStillAlive {
						// The watcher has been stopped at the client end, so we're
						// expecting one of the above two kinds of error.
						// We might see the same errors if the server itself
						// has been shut down, in which case we leave them
						// untouched.
						err = tomb.ErrDying
					}
				}
				// Something went wrong, just report the error and bail out.
				w.tomb.Kill(err)
				return
			}
			select {
			case <-w.tomb.Dying():
				return
			case w.in <- result:
				// Report back the result we just got.
			}
		}
	}()
	wg.Wait()
}
756442
func (m *Broadcaster) stopWatching(id int64) {
	m.lock.Lock()
	defer m.lock.Unlock()
	w, ok := m.watchers[id]
	if !ok {
		// No need to do anything, it's already been removed from the list.
		return
	}
	delete(m.watchers, id)
	close(w.result)
}
641929
func (r *runner) Stop() []error {
	var errs []error

	// Stop the monitor
	r.monitor.Stop()

	// TODO: Actually stop the plugins

	// For each delegate unregister needed handlers
	for _, del := range r.delegates {
		e := del.UnregisterHandler(HandlerRegistrationName)
		if e != nil {
			errs = append(errs, e)
		}
	}
	defer runnerLog.WithFields(log.Fields{
		"_block": "start-plugin",
	}).Debug("stopped")
	return errs
}
278565
func (w *Watcher) Close() error {
	w.closedMutex.Lock()
	defer w.closedMutex.Unlock()

	if w.isClosed {
		return nil
	}
	w.isClosed = true

	w.watchesMutex.Lock()
	for pid := range w.watches {
		delete(w.watches, pid)
		w.unregister(pid)
	}
	w.watchesMutex.Unlock()

	w.done <- true

	w.listener.close()

	return nil
}
1033107
func (t *TestTLSServer) Stop() error {
	err := t.TLSServer.Close()
	if t.Listener != nil {
		t.Listener.Close()
	}
	return err
}
276516
func (pw *ProxyWatcher) Stopping() bool {
	pw.mutex.Lock()
	defer pw.mutex.Unlock()
	return pw.stopped
}
1226039
func (r *Receiver) Stop() error {
	if !atomic.CompareAndSwapInt32(&r.state, stateListening, stateStopping) {
		return errors.New("nsq: cannot stop non-listening receiver")
	}

	r.cons.Stop()
	return nil
}
1112337
func (rpw *ReplicationWatcher) Close() {
	if !rpw.isOpen {
		return
	}
	rpw.cancel()
	rpw.wg.Wait()
	rpw.isOpen = false
}
374848
func (sw *StreamWatcher) Stop() {
	// Call Close() exactly once by locking and setting a flag.
	sw.Lock()
	defer sw.Unlock()
	if !sw.stopped {
		sw.stopped = true
		sw.source.Close()
	}
}
565900
func (fb *Firebase) StopWatching() {
	fb.watchMtx.Lock()
	defer fb.watchMtx.Unlock()

	if fb.watching {
		// flip the bit back to not watching
		fb.watching = false
		// signal connection to terminal
		fb.stopWatching <- struct{}{}
	}
}
858405
func (w *mockWatcher) Stop() {
	select {
	case <-w.stop:
		return
	default:
		close(w.stop)
		close(w.results)
	}
}
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
		}
	}()
}
1072267
func (g *GossipSyncer) Stop() {
	g.stopped.Do(func() {
		close(g.quit)
		g.wg.Wait()
	})
}