// condMonitor, returns a Monitor that broadcasts on cond on each update.
6806
func condMonitor(freq time.Duration, cond *sync.Cond) (*Monitor, error) {
	m := &Monitor{
		tick:   time.NewTicker(freq),
		inited: true,
		cond:   cond,
	}
	m.state.Set(stateRunning)
	if err := m.monitorLoop(); err != nil {
		return nil, err
	}
	return m, nil
}
1134180
func (m *Monitor) connectionHandler1_0(parentCtx context.Context, server net.Listener) {
	for !isCtxDone(parentCtx) {
		conn, err := server.Accept()
		switch {
		case isCtxDone(parentCtx) && conn != nil:
			conn.Close()
			fallthrough

		case isCtxDone(parentCtx) && conn == nil:
			return

		case err != nil:
			log.WithError(err).Warn("Error accepting connection")
			continue
		}

		m.registerNewListener(parentCtx, conn, listener.Version1_0)
	}
}
886890
func (m *MetricsServer) IncSync(app *argoappv1.Application, state *argoappv1.OperationState) {
	if !state.Phase.Completed() {
		return
	}
	m.syncCounter.WithLabelValues(app.Namespace, app.Name, app.Spec.GetProject(), string(state.Phase)).Inc()
}
673483
func (c *Client) monitorEvents() (EventMonitor, error) {
	if err := eventMonitor.run(c); err != nil {
		return nil, err
	}
	return eventMonitor, nil
}
310233
func (l *Listener) trackConn() func() {
	c := atomic.AddInt32(&l.activeConns, 1)
	metrics.SetGaugeWithLabels([]string{l.metricPrefix, "conns"}, float32(c),
		l.metricLabels)

	return func() {
		c := atomic.AddInt32(&l.activeConns, -1)
		metrics.SetGaugeWithLabels([]string{l.metricPrefix, "conns"}, float32(c),
			l.metricLabels)
	}
}
820314
func (w *Window) Monitor() *Monitor {
	var monitor *glfw.Monitor
	mainthread.Call(func() {
		monitor = w.window.GetMonitor()
	})
	if monitor == nil {
		return nil
	}
	return &Monitor{
		monitor: monitor,
	}
}
1140269
func SubOptionMonitorCNIExtractor(extractor extractors.EventMetadataExtractor) CNIMonitorOption {
	return func(cfg *cnimonitor.Config) {
		cfg.EventMetadataExtractor = extractor
	}
}
201925
func PrimaryMonitor() *Monitor {
	var monitor *glfw.Monitor
	mainthread.Call(func() {
		monitor = glfw.GetPrimaryMonitor()
	})
	return &Monitor{
		monitor: monitor,
	}
}
449711
func (b *BigIP) GetMonitor(name string, monitorType string) (*Monitor, error) {
	// Add a verification that type is an accepted monitor type
	var monitor Monitor
	err, ok := b.getForEntity(&monitor, uriLtm, uriMonitor, monitorType, name)
	if err != nil {
		return nil, err
	}
	if !ok {
		return nil, nil
	}

	return &monitor, nil
}
254682
func (c *Client) Conn() *grpc.ClientConn {
	c.connMu.Lock()
	defer c.connMu.Unlock()
	return c.conn
}
1092065
func (s *Swarm) ConnHandler() inet.ConnHandler {
	handler, _ := s.connh.Load().(inet.ConnHandler)
	return handler
}
894702
func (c *CreateCheckRunOptions) GetConclusion() string {
	if c == nil || c.Conclusion == nil {
		return ""
	}
	return *c.Conclusion
}
797785
func (s *Swarm) ConnHandler() ConnHandler {
	s.handlerLock.RLock()
	defer s.handlerLock.RUnlock()
	if s.connHandler == nil {
		return NoOpConnHandler
	}
	return s.connHandler
}
976530
func (client *Client) Monitor() {
	if err := recover(); err != nil {
		client.Notify(newError(err, 2))
		panic(err)
	}
}
569405
func (b *Broker) muCoordinatorConnection(consumerGroup string) (conn *connection, resErr error) {
	b.mu.Lock()
	defer b.mu.Unlock()

	for retry := 0; retry < b.conf.LeaderRetryLimit; retry++ {
		if retry != 0 {
			b.mu.Unlock()
			time.Sleep(b.conf.LeaderRetryWait)
			b.mu.Lock()
		}

		// first try all already existing connections
		for _, conn := range b.conns {
			resp, err := conn.ConsumerMetadata(&proto.ConsumerMetadataReq{
				RequestHeader: proto.RequestHeader{ClientID: b.conf.ClientID},
				ConsumerGroup: consumerGroup,
			})
			if err != nil {
				b.conf.Logger.Debug("cannot fetch coordinator metadata",
					"consumGrp", consumerGroup,
					"error", err)
				resErr = err
				continue
			}
			if resp.Err != nil {
				b.conf.Logger.Debug("coordinator metadata response error",
					"consumGrp", consumerGroup,
					"error", resp.Err)
				resErr = err
				continue
			}

			addr := fmt.Sprintf("%s:%d", resp.CoordinatorHost, resp.CoordinatorPort)
			conn, err := b.getConnection(addr)
			if err != nil {
				b.conf.Logger.Debug("cannot connect to node",
					"coordinatorID", resp.CoordinatorID,
					"address", addr,
					"error", err)
				resErr = err
				continue
			}
			b.conns[resp.CoordinatorID] = conn
			return conn, nil
		}

		// if none of the connections worked out, try with fresh data
		if err := b.refreshMetadata(); err != nil {
			b.conf.Logger.Debug("cannot refresh metadata",
				"error", err)
			resErr = err
			continue
		}

		for nodeID, addr := range b.metadata.nodes {
			if _, ok := b.conns[nodeID]; ok {
				// connection to node is cached so it was already checked
				continue
			}
			conn, err := b.getConnection(addr)
			if err != nil {
				b.conf.Logger.Debug("cannot connect to node",
					"nodeID", nodeID,
					"address", addr,
					"error", err)
				resErr = err
				continue
			}
			b.conns[nodeID] = conn

			resp, err := conn.ConsumerMetadata(&proto.ConsumerMetadataReq{
				RequestHeader: proto.RequestHeader{ClientID: b.conf.ClientID},
				ConsumerGroup: consumerGroup,
			})
			if err != nil {
				b.conf.Logger.Debug("cannot fetch metadata",
					"consumGrp", consumerGroup,
					"error", err)
				resErr = err
				continue
			}
			if resp.Err != nil {
				b.conf.Logger.Debug("metadata response error",
					"consumGrp", consumerGroup,
					"error", resp.Err)
				resErr = err
				continue
			}

			addr := fmt.Sprintf("%s:%d", resp.CoordinatorHost, resp.CoordinatorPort)
			conn, err = b.getConnection(addr)
			if err != nil {
				b.conf.Logger.Debug("cannot connect to node",
					"coordinatorID", resp.CoordinatorID,
					"address", addr,
					"error", err)
				resErr = err
				continue
			}
			b.conns[resp.CoordinatorID] = conn
			return conn, nil
		}
		resErr = proto.ErrNoCoordinator
	}
	return nil, resErr
}
990910
func (s *Server) Connz(opts *ConnzOptions) (*Connz, error) {
	var (
		sortOpt = ByCid
		auth    bool
		subs    bool
		offset  int
		limit   = DefaultConnListSize
		cid     = uint64(0)
		state   = ConnOpen
	)

	if opts != nil {
		// If no sort option given or sort is by uptime, then sort by cid
		if opts.Sort == "" {
			sortOpt = ByCid
		} else {
			sortOpt = opts.Sort
			if !sortOpt.IsValid() {
				return nil, fmt.Errorf("invalid sorting option: %s", sortOpt)
			}
		}
		auth = opts.Username
		subs = opts.Subscriptions
		offset = opts.Offset
		if offset < 0 {
			offset = 0
		}
		limit = opts.Limit
		if limit <= 0 {
			limit = DefaultConnListSize
		}
		// state
		state = opts.State

		// ByStop only makes sense on closed connections
		if sortOpt == ByStop && state != ConnClosed {
			return nil, fmt.Errorf("sort by stop only valid on closed connections")
		}
		// ByReason is the same.
		if sortOpt == ByReason && state != ConnClosed {
			return nil, fmt.Errorf("sort by reason only valid on closed connections")
		}

		// If searching by CID
		if opts.CID > 0 {
			cid = opts.CID
			limit = 1
		}
	}

	c := &Connz{
		Offset: offset,
		Limit:  limit,
		Now:    time.Now(),
	}

	// Open clients
	var openClients []*client
	// Hold for closed clients if requested.
	var closedClients []*closedClient

	// Walk the open client list with server lock held.
	s.mu.Lock()

	// copy the server id for monitoring
	c.ID = s.info.ID

	// Number of total clients. The resulting ConnInfo array
	// may be smaller if pagination is used.
	switch state {
	case ConnOpen:
		c.Total = len(s.clients)
	case ConnClosed:
		c.Total = s.closed.len()
		closedClients = s.closed.closedClients()
		c.Total = len(closedClients)
	case ConnAll:
		closedClients = s.closed.closedClients()
		c.Total = len(s.clients) + len(closedClients)
	}

	totalClients := c.Total
	if cid > 0 { // Meaning we only want 1.
		totalClients = 1
	}
	if state == ConnOpen || state == ConnAll {
		openClients = make([]*client, 0, totalClients)
	}

	// Data structures for results.
	var conns []ConnInfo // Limits allocs for actual ConnInfos.
	var pconns ConnInfos

	switch state {
	case ConnOpen:
		conns = make([]ConnInfo, totalClients)
		pconns = make(ConnInfos, totalClients)
	case ConnClosed:
		pconns = make(ConnInfos, totalClients)
	case ConnAll:
		conns = make([]ConnInfo, cap(openClients))
		pconns = make(ConnInfos, totalClients)
	}

	// Search by individual CID.
	if cid > 0 {
		if state == ConnClosed || state == ConnAll {
			copyClosed := closedClients
			closedClients = nil
			for _, cc := range copyClosed {
				if cc.Cid == cid {
					closedClients = []*closedClient{cc}
					break
				}
			}
		} else if state == ConnOpen || state == ConnAll {
			client := s.clients[cid]
			if client != nil {
				openClients = append(openClients, client)
			}
		}
	} else {
		// Gather all open clients.
		if state == ConnOpen || state == ConnAll {
			for _, client := range s.clients {
				openClients = append(openClients, client)
			}
		}
	}
	s.mu.Unlock()

	// Just return with empty array if nothing here.
	if len(openClients) == 0 && len(closedClients) == 0 {
		c.Conns = ConnInfos{}
		return c, nil
	}

	// Now whip through and generate ConnInfo entries

	// Open Clients
	i := 0
	for _, client := range openClients {
		client.mu.Lock()
		ci := &conns[i]
		ci.fill(client, client.nc, c.Now)
		// Fill in subscription data if requested.
		if subs && len(client.subs) > 0 {
			ci.Subs = make([]string, 0, len(client.subs))
			for _, sub := range client.subs {
				ci.Subs = append(ci.Subs, string(sub.subject))
			}
		}
		// Fill in user if auth requested.
		if auth {
			ci.AuthorizedUser = client.opts.Username
		}
		client.mu.Unlock()
		pconns[i] = ci
		i++
	}
	// Closed Clients
	var needCopy bool
	if subs || auth {
		needCopy = true
	}
	for _, cc := range closedClients {
		// Copy if needed for any changes to the ConnInfo
		if needCopy {
			cx := *cc
			cc = &cx
		}
		// Fill in subscription data if requested.
		if subs && len(cc.subs) > 0 {
			cc.Subs = cc.subs
		}
		// Fill in user if auth requested.
		if auth {
			cc.AuthorizedUser = cc.user
		}
		pconns[i] = &cc.ConnInfo
		i++
	}

	switch sortOpt {
	case ByCid, ByStart:
		sort.Sort(byCid{pconns})
	case BySubs:
		sort.Sort(sort.Reverse(bySubs{pconns}))
	case ByPending:
		sort.Sort(sort.Reverse(byPending{pconns}))
	case ByOutMsgs:
		sort.Sort(sort.Reverse(byOutMsgs{pconns}))
	case ByInMsgs:
		sort.Sort(sort.Reverse(byInMsgs{pconns}))
	case ByOutBytes:
		sort.Sort(sort.Reverse(byOutBytes{pconns}))
	case ByInBytes:
		sort.Sort(sort.Reverse(byInBytes{pconns}))
	case ByLast:
		sort.Sort(sort.Reverse(byLast{pconns}))
	case ByIdle:
		sort.Sort(sort.Reverse(byIdle{pconns}))
	case ByUptime:
		sort.Sort(byUptime{pconns, time.Now()})
	case ByStop:
		sort.Sort(sort.Reverse(byStop{pconns}))
	case ByReason:
		sort.Sort(byReason{pconns})
	}

	minoff := c.Offset
	maxoff := c.Offset + c.Limit

	maxIndex := totalClients

	// Make sure these are sane.
	if minoff > maxIndex {
		minoff = maxIndex
	}
	if maxoff > maxIndex {
		maxoff = maxIndex
	}

	// Now pare down to the requested size.
	// TODO(dlc) - for very large number of connections we
	// could save the whole list in a hash, send hash on first
	// request and allow users to use has for subsequent pages.
	// Low TTL, say < 1sec.
	c.Conns = pconns[minoff:maxoff]
	c.NumConns = len(c.Conns)

	return c, nil
}
162322
func BuilderConnector(b *zkplus.Builder) ZkConnCreator {
	return ZkConnCreatorFunc(func() (ZkConn, <-chan zk.Event, error) {
		return b.BuildDirect()
	})
}