// Run executes the Bee's event loop.
12801
func (mod *IrcBee) Run(eventChan chan bees.Event) {
	if len(mod.server) == 0 {
		return
	}

	// channel signaling IRC connection status
	mod.connectedState = make(chan bool)

	// setup IRC client:
	cfg := irc.NewConfig(mod.nick, "beehive", "beehive")
	cfg.SSL = mod.ssl
	if mod.ssl {
		h, _, _ := net.SplitHostPort(mod.server)
		if h == "" {
			h = mod.server
		}
		cfg.SSLConfig = &tls.Config{ServerName: h}
	}

	cfg.Server = mod.server
	cfg.Pass = mod.password
	cfg.NewNick = func(n string) string { return n + "_" }
	mod.client = irc.Client(cfg)

	mod.client.HandleFunc("connected", func(conn *irc.Conn, line *irc.Line) {
		mod.connectedState <- true
	})
	mod.client.HandleFunc("disconnected", func(conn *irc.Conn, line *irc.Line) {
		mod.connectedState <- false
	})

	mod.client.HandleFunc("JOIN", func(conn *irc.Conn, line *irc.Line) {
		mod.statusChange(eventChan, conn, line)
	})
	mod.client.HandleFunc("PART", func(conn *irc.Conn, line *irc.Line) {
		mod.statusChange(eventChan, conn, line)
	})
	mod.client.HandleFunc("QUIT", func(conn *irc.Conn, line *irc.Line) {
		mod.statusChange(eventChan, conn, line)
	})

	mod.client.HandleFunc("PRIVMSG", func(conn *irc.Conn, line *irc.Line) {
		channel := line.Args[0]
		if channel == mod.client.Config().Me.Nick {
			channel = line.Src // replies go via PM too
		}
		msg := ""
		if len(line.Args) > 1 {
			msg = line.Args[1]
		}
		user := line.Src[:strings.Index(line.Src, "!")]
		hostmask := line.Src[strings.Index(line.Src, "!")+2:]

		ev := bees.Event{
			Bee:  mod.Name(),
			Name: "message",
			Options: []bees.Placeholder{
				{
					Name:  "channel",
					Type:  "string",
					Value: channel,
				},
				{
					Name:  "user",
					Type:  "string",
					Value: user,
				},
				{
					Name:  "hostmask",
					Type:  "string",
					Value: hostmask,
				},
				{
					Name:  "text",
					Type:  "string",
					Value: msg,
				},
			},
		}
		eventChan <- ev
	})

	waitForDisconnect := false
	connecting := false
	connected := false
	mod.ContextSet("connected", &connected)

	for {
		// loop on IRC connection events
		if !connected {
			if waitForDisconnect {
				return
			}

			if !connecting {
				connecting = true
				mod.Logln("Connecting to IRC:", mod.server)
				err := mod.client.Connect()
				if err != nil {
					mod.LogErrorf("Failed to connect to IRC: %s %v", mod.server, err)
					connecting = false
				}
			}
		}
		select {
		case status := <-mod.connectedState:
			if status {
				mod.Logln("Connected to IRC:", mod.server)
				connecting = false
				connected = true
				mod.rejoin()
			} else {
				mod.Logln("Disconnected from IRC:", mod.server)
				connecting = false
				connected = false
			}

		case <-mod.SigChan:
			if !waitForDisconnect {
				mod.client.Quit()
			}
			waitForDisconnect = true

		default:
			time.Sleep(time.Second)
		}
	}
}
946759
func (s *Server) Run() error {
	metrics.ServerEventCounter.WithLabelValues(metrics.EventStart).Inc()

	// Start HTTP API to report tidb info such as TPS.
	if s.cfg.Status.ReportStatus {
		s.startStatusHTTP()
	}
	for {
		conn, err := s.listener.Accept()
		if err != nil {
			if opErr, ok := err.(*net.OpError); ok {
				if opErr.Err.Error() == "use of closed network connection" {
					return nil
				}
			}

			// If we got PROXY protocol error, we should continue accept.
			if proxyprotocol.IsProxyProtocolError(err) {
				logutil.Logger(context.Background()).Error("PROXY protocol failed", zap.Error(err))
				continue
			}

			logutil.Logger(context.Background()).Error("accept failed", zap.Error(err))
			return errors.Trace(err)
		}
		if s.shouldStopListener() {
			err = conn.Close()
			terror.Log(errors.Trace(err))
			break
		}

		clientConn := s.newConn(conn)

		err = plugin.ForeachPlugin(plugin.Audit, func(p *plugin.Plugin) error {
			authPlugin := plugin.DeclareAuditManifest(p.Manifest)
			if authPlugin.OnConnectionEvent != nil {
				host, err := clientConn.PeerHost("")
				if err != nil {
					logutil.Logger(context.Background()).Error("get peer host failed", zap.Error(err))
					terror.Log(clientConn.Close())
					return errors.Trace(err)
				}
				err = authPlugin.OnConnectionEvent(context.Background(), &auth.UserIdentity{Hostname: host}, plugin.PreAuth, nil)
				if err != nil {
					logutil.Logger(context.Background()).Info("do connection event failed", zap.Error(err))
					terror.Log(clientConn.Close())
					return errors.Trace(err)
				}
			}
			return nil
		})
		if err != nil {
			continue
		}

		go s.onConn(clientConn)
	}
	err := s.listener.Close()
	terror.Log(errors.Trace(err))
	s.listener = nil
	for {
		metrics.ServerEventCounter.WithLabelValues(metrics.EventHang).Inc()
		logutil.Logger(context.Background()).Error("listener stopped, waiting for manual kill.")
		time.Sleep(time.Minute)
	}
}
42805
func (mod *TimeBee) Run(eventChan chan bees.Event) {
	mod.eventChan = eventChan
	for {
		select {
		case <-mod.SigChan:
			return

		default:
		}

		mod.timer()
		time.Sleep(500 * time.Millisecond)
	}
}
12264
func (c *CloudNodeLifecycleController) Run(stopCh <-chan struct{}) {
	defer utilruntime.HandleCrash()

	// The following loops run communicate with the APIServer with a worst case complexity
	// of O(num_nodes) per cycle. These functions are justified here because these events fire
	// very infrequently. DO NOT MODIFY this to perform frequent operations.

	// Start a loop to periodically check if any nodes have been
	// deleted or shutdown from the cloudprovider
	wait.Until(c.MonitorNodes, c.nodeMonitorPeriod, stopCh)
}
460148
func (mod *HorizonBoxBee) Run(cin chan bees.Event) {
	mod.eventChan = cin
	for {
		select {
		case <-mod.SigChan:
			return

		case <-time.After(time.Duration(mod.interval) * time.Second):
			mod.poll()
		}
	}
}
27465
func (mod *PrometheusBee) Run(eventChan chan bees.Event) {

	// Counter vector registration

	mod.counter = prometheus.NewCounterVec(
		prometheus.CounterOpts{
			Name: mod.counterVecName,
			Help: "Collection of all counter variables being iterated by hives",
		},
		[]string{"name"},
	)
	err := prometheus.Register(mod.counter)
	if err != nil {
		mod.LogErrorf("Error registering counter vector: %v", err)
		panic("Unable to start Prometheus due to counter initialization failure")
	}

	// Gauge vector registration

	mod.gauge = prometheus.NewGaugeVec(
		prometheus.GaugeOpts{
			Name: mod.gaugeVecName,
			Help: "Collection of all gauge variables being iterated by hives",
		},
		[]string{"name"},
	)
	err = prometheus.Register(mod.gauge)
	if err != nil {
		mod.LogErrorf("Error registering gauge vector: %v", err)
		panic("Unable to start Prometheus due to gauge initialization failure")
	}

	// Histogram vector registration

	mod.histogram = prometheus.NewHistogramVec(
		prometheus.HistogramOpts{
			Name: mod.histogramVecName,
			Help: "Collection of all histogram variables being iterated by hives",
		},
		[]string{"name"},
	)
	err = prometheus.Register(mod.histogram)
	if err != nil {
		mod.LogErrorf("Error registering histogram vector: %v", err)
		panic("Unable to start Prometheus due to histogram initialization failure")
	}

	// Summary vector registration

	mod.summary = prometheus.NewSummaryVec(
		prometheus.SummaryOpts{
			Name: mod.summaryVecName,
			Help: "Collection of all summary variables being iterated by hives",
		},
		[]string{"name"},
	)
	err = prometheus.Register(mod.summary)
	if err != nil {
		mod.LogErrorf("Error registering summary vector: %v", err)
		panic("Unable to start Prometheus due to summary initialization failure")
	}

	// Now, to serve everything up:

	go func() {
		http.Handle("/metrics", promhttp.Handler())
		err := http.ListenAndServe(":2112", nil)
		if err != nil {
			mod.LogErrorf("Error running prometheus metric handler: %v", err)
		}
	}()

	select {
	case <-mod.SigChan:
		return
	}
}
148778
func (t *Timer) Run() {
	log.Print("Timer is running now.")
	go func() {
		for {
			select {
			case <-t.ticker.C:
				go func() {
					log.Print("Worker is doing its jobs.")
					t.worker.Work()
				}()
			}
		}
	}()
}
968057
func (mod *SlackBee) Run(eventChan chan bees.Event) {
	rtm := mod.client.NewRTM()
	defer rtm.Disconnect()

	go rtm.ManageConnection()

	// Cache the channels
	for k := range mod.channels {
		mod.findChannelID(k, true)
	}

	for {
		select {
		case <-mod.SigChan:
			return

		case msg := <-rtm.IncomingEvents:
			switch ev := msg.Data.(type) {
			case *slack.MessageEvent:
				if stringInMap(ev.Channel, mod.channels) {
					u := ev.Msg.User
					if u == "" {
						u = ev.Msg.Username
					}
					t := ev.Msg.Text
					if t == "" {
						for _, v := range ev.Msg.Attachments {
							sendEvent(mod.Name(), ev.Channel, u, v.Text, eventChan)
						}
					} else {
						sendEvent(mod.Name(), ev.Channel, u, t, eventChan)
					}
				}
			case *slack.RTMError:
				mod.LogErrorf("Error: %s", ev.Error())
			case *slack.InvalidAuthEvent:
				mod.LogErrorf("Error: invalid credentials")
				return
			default:
				// Ignore other events..
				// fmt.Printf("Unexpected: %v\n", msg.Data)
			}
		}
	}
}
1068337
func (u *DropsondeUnmarshallerCollection) Run(inputChan <-chan []byte, outputChan chan<- *events.Envelope, waitGroup *sync.WaitGroup) {
	for _, unmarshaller := range u.unmarshallers {
		go func(um *DropsondeUnmarshaller) {
			defer waitGroup.Done()
			um.Run(inputChan, outputChan)
		}(unmarshaller)
	}
}
972710
func (mod *FacebookBee) Run(eventChan chan bees.Event) {
	mod.evchan = eventChan

	since := strconv.FormatInt(time.Now().UTC().Unix(), 10)
	timeout := time.Duration(10 * time.Second)

	for {
		select {
		case <-mod.SigChan:
			return
		case <-time.After(time.Duration(timeout)):
			var err error
			since, err = mod.handleStream(since)
			if err != nil {
				panic(err)
			}
		}

		timeout = time.Minute
	}
}
835857
func (w *checkWatcher) Run(ctx context.Context) {
	defer close(w.done)

	// map of check IDs to their metadata
	checks := map[string]*checkRestart{}

	// timer for check polling
	checkTimer := time.NewTimer(0)
	defer checkTimer.Stop() // ensure timer is never leaked

	stopTimer := func() {
		checkTimer.Stop()
		select {
		case <-checkTimer.C:
		default:
		}
	}

	// disable by default
	stopTimer()

	// Main watch loop
	for {
		// disable polling if there are no checks
		if len(checks) == 0 {
			stopTimer()
		}

		select {
		case update := <-w.checkUpdateCh:
			if update.remove {
				// Remove a check
				delete(checks, update.checkID)
				continue
			}

			// Add/update a check
			checks[update.checkID] = update.checkRestart
			w.logger.Debug("watching check", "alloc_id", update.checkRestart.allocID,
				"task", update.checkRestart.taskName, "check", update.checkRestart.checkName)

			// if first check was added make sure polling is enabled
			if len(checks) == 1 {
				stopTimer()
				checkTimer.Reset(w.pollFreq)
			}

		case <-ctx.Done():
			return

		case <-checkTimer.C:
			checkTimer.Reset(w.pollFreq)

			// Set "now" as the point in time the following check results represent
			now := time.Now()

			results, err := w.consul.Checks()
			if err != nil {
				if !w.lastErr {
					w.lastErr = true
					w.logger.Error("failed retrieving health checks", "error", err)
				}
				continue
			}

			w.lastErr = false

			// Keep track of tasks restarted this period so they
			// are only restarted once and all of their checks are
			// removed.
			restartedTasks := map[string]struct{}{}

			// Loop over watched checks and update their status from results
			for cid, check := range checks {
				// Shortcircuit if told to exit
				if ctx.Err() != nil {
					return
				}

				if _, ok := restartedTasks[check.taskKey]; ok {
					// Check for this task already restarted; remove and skip check
					delete(checks, cid)
					continue
				}

				result, ok := results[cid]
				if !ok {
					// Only warn if outside grace period to avoid races with check registration
					if now.After(check.graceUntil) {
						w.logger.Warn("watched check not found in Consul", "check", check.checkName, "check_id", cid)
					}
					continue
				}

				restarted := check.apply(ctx, now, result.Status)
				if restarted {
					// Checks are registered+watched on
					// startup, so it's safe to remove them
					// whenever they're restarted
					delete(checks, cid)

					restartedTasks[check.taskKey] = struct{}{}
				}
			}

			// Ensure even passing checks for restartedTasks are removed
			if len(restartedTasks) > 0 {
				for cid, check := range checks {
					if _, ok := restartedTasks[check.taskKey]; ok {
						delete(checks, cid)
					}
				}
			}
		}
	}
}
95643
func (j *resizeJob) run() error {
	j.Logger.Printf("run resizeJob")
	// Set job state to RUNNING.
	j.setState(resizeJobStateRunning)

	// Job can be considered done in the case where it doesn't require any action.
	if !j.nodesArePending() {
		j.Logger.Printf("resizeJob contains no pending tasks; mark as done")
		j.result <- resizeJobStateDone
		return nil
	}

	j.Logger.Printf("distribute tasks for resizeJob")
	err := j.distributeResizeInstructions()
	if err != nil {
		j.result <- resizeJobStateAborted
		return errors.Wrap(err, "distributing instructions")
	}
	return nil
}
349001
func (b *Broadcaster) run() {
	for {
		select {
		case block := <-b.events:
			for _, sink := range b.sinks {
				if err := sink.Write(block...); err != nil {
					logrus.Errorf("broadcaster: error writing events to %v, these events will be lost: %v", sink, err)
				}
			}
		case closing := <-b.closed:

			// close all the underlying sinks
			for _, sink := range b.sinks {
				if err := sink.Close(); err != nil {
					logrus.Errorf("broadcaster: error closing sink %v: %v", sink, err)
				}
			}
			closing <- struct{}{}

			logrus.Debugf("broadcaster: closed")
			return
		}
	}
}
566186
func (mod *DevrantBee) Run(eventChan chan bees.Event) {
	mod.evchan = eventChan

	// Setting up the client, unfortunately we can't really log in as an user
	mod.client = devgorant.New()
}
1023509
func (mod *TumblrBee) Run(eventChan chan bees.Event) {
	mod.evchan = eventChan

	select {
	case <-mod.SigChan:
		return
	}
}
461957
func (a *AllocGarbageCollector) Run() {
	ticker := time.NewTicker(a.config.Interval)
	for {
		select {
		case <-a.triggerCh:
		case <-ticker.C:
		case <-a.shutdownCh:
			ticker.Stop()
			return
		}

		if err := a.keepUsageBelowThreshold(); err != nil {
			a.logger.Error("error garbage collecting allocations", "error", err)
		}
	}
}
695468
func (c *Cron) Run() {
	if c.running {
		return
	}
	c.running = true
	c.run()
}