// Collect registers the container with the collector and adds it to
// the event loop for collection on the specified interval returning
// a channel for the subscriber to receive on.
488068
func (s *Collector) Collect(c *container.Container) chan interface{} {
	s.m.Lock()
	defer s.m.Unlock()
	publisher, exists := s.publishers[c]
	if !exists {
		publisher = pubsub.NewPublisher(100*time.Millisecond, 1024)
		s.publishers[c] = publisher
	}
	return publisher.Subscribe()
}
554430
func (m *ServerMetrics) Collect(ch chan<- prom.Metric) {
	m.serverStartedCounter.Collect(ch)
	m.serverHandledCounter.Collect(ch)
	m.serverStreamMsgReceived.Collect(ch)
	m.serverStreamMsgSent.Collect(ch)
	if m.serverHandledHistogramEnabled {
		m.serverHandledHistogram.Collect(ch)
	}
}
118589
func (a *Group) Collect(waitables ...Waitable) {
	a.Lock()
	a.waitables = append(a.waitables, waitables...)
	a.Unlock()
}
1281082
func (m *MockGauge) Collect(arg0 chan<- prometheus.Metric) {
	m.ctrl.Call(m, "Collect", arg0)
}
341770
func (c *ImportMetricCounter) Collect() (metrics.ImportSuccessCounts, metrics.ImportErrorCounts, error) {
	c.counterMutex.Lock()
	defer c.counterMutex.Unlock()

	success := metrics.ImportSuccessCounts{}
	for registry, count := range c.importSuccessCounts {
		success[registry] = count
	}

	failures := metrics.ImportErrorCounts{}
	for info, count := range c.importErrorCounts {
		failures[info] = count
	}

	return success, failures, nil
}
998466
func (m *MockCollector) CollectTraceEvent(records []string) {
	m.ctrl.Call(m, "CollectTraceEvent", records)
}
400813
func (s *UtxoSweeper) collector(blockEpochs <-chan *chainntnfs.BlockEpoch,
	bestHeight int32) error {

	for {
		select {
		// A new inputs is offered to the sweeper. We check to see if we
		// are already trying to sweep this input and if not, set up a
		// listener for spend and schedule a sweep.
		case input := <-s.newInputs:
			outpoint := *input.input.OutPoint()
			pendInput, pending := s.pendingInputs[outpoint]
			if pending {
				log.Debugf("Already pending input %v received",
					outpoint)

				// Add additional result channel to signal
				// spend of this input.
				pendInput.listeners = append(
					pendInput.listeners, input.resultChan,
				)
				continue
			}

			// Create a new pendingInput and initialize the
			// listeners slice with the passed in result channel. If
			// this input is offered for sweep again, the result
			// channel will be appended to this slice.
			pendInput = &pendingInput{
				listeners:        []chan Result{input.resultChan},
				input:            input.input,
				minPublishHeight: bestHeight,
			}
			s.pendingInputs[outpoint] = pendInput

			// Start watching for spend of this input, either by us
			// or the remote party.
			cancel, err := s.waitForSpend(
				outpoint,
				input.input.SignDesc().Output.PkScript,
				input.input.HeightHint(),
			)
			if err != nil {
				err := fmt.Errorf("wait for spend: %v", err)
				s.signalAndRemove(&outpoint, Result{Err: err})
				continue
			}
			pendInput.ntfnRegCancel = cancel

			// Check to see if with this new input a sweep tx can be
			// formed.
			if err := s.scheduleSweep(bestHeight); err != nil {
				log.Errorf("schedule sweep: %v", err)
			}

		// A spend of one of our inputs is detected. Signal sweep
		// results to the caller(s).
		case spend := <-s.spendChan:
			// For testing purposes.
			if s.testSpendChan != nil {
				s.testSpendChan <- *spend.SpentOutPoint
			}

			// Query store to find out if we every published this
			// tx.
			spendHash := *spend.SpenderTxHash
			isOurTx, err := s.cfg.Store.IsOurTx(spendHash)
			if err != nil {
				log.Errorf("cannot determine if tx %v "+
					"is ours: %v", spendHash, err,
				)
				continue
			}

			log.Debugf("Detected spend related to in flight inputs "+
				"(is_ours=%v): %v",
				newLogClosure(func() string {
					return spew.Sdump(spend.SpendingTx)
				}), isOurTx,
			)

			// Signal sweep results for inputs in this confirmed
			// tx.
			for _, txIn := range spend.SpendingTx.TxIn {
				outpoint := txIn.PreviousOutPoint

				// Check if this input is known to us. It could
				// probably be unknown if we canceled the
				// registration, deleted from pendingInputs but
				// the ntfn was in-flight already. Or this could
				// be not one of our inputs.
				_, ok := s.pendingInputs[outpoint]
				if !ok {
					continue
				}

				// Return either a nil or a remote spend result.
				var err error
				if !isOurTx {
					err = ErrRemoteSpend
				}

				// Signal result channels.
				s.signalAndRemove(&outpoint, Result{
					Tx:  spend.SpendingTx,
					Err: err,
				})
			}

			// Now that an input of ours is spent, we can try to
			// resweep the remaining inputs.
			if err := s.scheduleSweep(bestHeight); err != nil {
				log.Errorf("schedule sweep: %v", err)
			}

		// The timer expires and we are going to (re)sweep.
		case <-s.timer:
			log.Debugf("Sweep timer expired")

			// Set timer to nil so we know that a new timer needs to
			// be started when new inputs arrive.
			s.timer = nil

			// Retrieve fee estimate for input filtering and final
			// tx fee calculation.
			satPerKW, err := s.cfg.FeeEstimator.EstimateFeePerKW(
				s.cfg.SweepTxConfTarget,
			)
			if err != nil {
				log.Errorf("estimate fee: %v", err)
				continue
			}

			// Examine pending inputs and try to construct lists of
			// inputs.
			inputLists, err := s.getInputLists(bestHeight, satPerKW)
			if err != nil {
				log.Errorf("get input lists: %v", err)
				continue
			}

			// Sweep selected inputs.
			for _, inputs := range inputLists {
				err := s.sweep(inputs, satPerKW, bestHeight)
				if err != nil {
					log.Errorf("sweep: %v", err)
				}
			}

		// A new block comes in. Things may have changed, so we retry a
		// sweep.
		case epoch, ok := <-blockEpochs:
			if !ok {
				return nil
			}

			bestHeight = epoch.Height

			log.Debugf("New blocks: height=%v, sha=%v",
				epoch.Height, epoch.Hash)

			if err := s.scheduleSweep(bestHeight); err != nil {
				log.Errorf("schedule sweep: %v", err)
			}

		case <-s.quit:
			return nil
		}
	}
}
40087
func (c *KafkaCollector) Collect(s *zipkincore.Span) error {
	c.producer.Input() <- &sarama.ProducerMessage{
		Topic: c.topic,
		Key:   nil,
		Value: sarama.ByteEncoder(kafkaSerialize(s)),
	}
	return nil
}
189506
func (h *StatsHandler) Collect(in chan<- prometheus.Metric) {
	for _, c := range h.handlers {
		c.Collect(in)
	}
}
1180129
func (c *MgoStatsCollector) Collect(ch chan<- prometheus.Metric) {
	stats, prevStats := c.getStats()

	c.clustersGauge.Set(float64(stats.Clusters))
	c.masterConnsGauge.Set(float64(stats.MasterConns))
	c.slaveConnsGauge.Set(float64(stats.SlaveConns))
	if n := stats.SentOps - prevStats.SentOps; n > 0 {
		c.sentOpsCounter.Add(float64(n))
	}
	if n := stats.ReceivedOps - prevStats.ReceivedOps; n > 0 {
		c.receivedOpsCounter.Add(float64(n))
	}
	if n := stats.ReceivedDocs - prevStats.ReceivedDocs; n > 0 {
		c.receivedDocsCounter.Add(float64(n))
	}
	c.socketsAliveGauge.Set(float64(stats.SocketsAlive))
	c.socketsInuseGauge.Set(float64(stats.SocketsInUse))
	c.socketRefsGauge.Set(float64(stats.SocketRefs))

	c.clustersGauge.Collect(ch)
	c.masterConnsGauge.Collect(ch)
	c.slaveConnsGauge.Collect(ch)
	c.sentOpsCounter.Collect(ch)
	c.receivedOpsCounter.Collect(ch)
	c.receivedDocsCounter.Collect(ch)
	c.socketsAliveGauge.Collect(ch)
	c.socketsInuseGauge.Collect(ch)
	c.socketRefsGauge.Collect(ch)
}
276406
func (c CollectorF) Collect(key, val SophieWriter) error {
	if c == nil {
		return nil
	}
	return c(key, val)
}
540783
func (c *influxCollector) Collect(ch chan<- prometheus.Metric) {
	ch <- prometheus.MustNewConstMetric(c.influxInfoDesc, prometheus.GaugeValue, 1)

	uptime := time.Since(c.start).Seconds()
	ch <- prometheus.MustNewConstMetric(c.influxUptimeDesc, prometheus.GaugeValue, float64(uptime))
}
957090
func (n *NeptuneApex) Gather(acc telegraf.Accumulator) error {
	var wg sync.WaitGroup
	for _, server := range n.Servers {
		wg.Add(1)
		go func(server string) {
			defer wg.Done()
			acc.AddError(n.gatherServer(acc, server))
		}(server)
	}
	wg.Wait()
	return nil
}
346021
func (c *timingsCollector) Collect(ch chan<- prometheus.Metric) {
	for cat, his := range c.t.Histograms() {
		ch <- prometheus.MustNewConstHistogram(
			c.desc,
			uint64(his.Count()),
			float64(his.Total())/1000000000,
			makeCumulativeBuckets(c.cutoffs, his.Buckets()),
			cat)
	}
}
1140979
func (mr *MockEventCollectorMockRecorder) CollectTraceEvent(records interface{}) *gomock.Call {
	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "CollectTraceEvent", reflect.TypeOf((*MockEventCollector)(nil).CollectTraceEvent), records)
}
216741
func (c *appCollector) Collect(ch chan<- prometheus.Metric) {
	apps, err := c.store.List(labels.NewSelector())
	if err != nil {
		log.Warnf("Failed to collect applications: %v", err)
		return
	}
	for _, app := range apps {
		collectApps(ch, app)
	}
}
1327715
func (c *metricsFuncWithMultiLabelsCollector) Collect(ch chan<- prometheus.Metric) {
	for lvs, val := range c.cfml.Counts() {
		labelValues := strings.Split(lvs, ".")
		value := float64(val)
		ch <- prometheus.MustNewConstMetric(c.desc, c.vt, value, labelValues...)
	}
}