// run is the main broadcast loop, started when the broadcaster is created.
// Under normal conditions, it waits for events on the event channel. After
// Close is called, this goroutine will exit.
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
		}
	}
}
399053
func (s *subscriber) run(hdlr broker.Handler) {
	log.Log(fmt.Sprintf("SQS subscription started. Queue:%s, URL: %s", s.queueName, s.URL))
	for {
		select {
		case <-s.exit:
			return
		default:
			result, err := s.svc.ReceiveMessage(&sqs.ReceiveMessageInput{
				QueueUrl:            &s.URL,
				MaxNumberOfMessages: s.getMaxMessages(),
				VisibilityTimeout:   s.getVisibilityTimeout(),
				WaitTimeSeconds:     s.getWaitSeconds(),
				AttributeNames: aws.StringSlice([]string{
					"SentTimestamp", // TODO: not currently exposing this to plugin users
				}),
				MessageAttributeNames: aws.StringSlice([]string{
					"All",
				}),
			})

			if err != nil {
				time.Sleep(time.Second)
				log.Log(fmt.Sprintf("Error receiving SQS message: %s", err.Error()))
				continue
			}

			if len(result.Messages) == 0 {
				time.Sleep(time.Second)
				continue
			}

			for _, sm := range result.Messages {
				s.handleMessage(sm, hdlr)
			}
		}
	}
}
475336
func (b *broker) run() error {
	listener, err := net.Listen("tcp", *addr)
	if err != nil {
		return err
	}
	defer listener.Close()
	fmt.Printf("Listening on %s\n", listener.Addr())
	for {
		conn, err := listener.Accept()
		if err != nil {
			debugf("Accept error: %v", err)
			continue
		}
		adapter := proton.NewMessagingAdapter(newHandler(&b.queues))
		// We want to accept messages when they are enqueued, not just when they
		// are received, so we turn off auto-accept and prefetch by the adapter.
		adapter.Prefetch = 0
		adapter.AutoAccept = false
		engine, err := proton.NewEngine(conn, adapter)
		if err != nil {
			debugf("Connection error: %v", err)
			continue
		}
		engine.Server() // Enable server-side protocol negotiation.
		debugf("Accepted connection %s", engine)
		go func() { // Start goroutine to run the engine event loop
			engine.Run()
			debugf("Closed %s", engine)
		}()
	}
}
1311536
func (s *stateSync) run() {
	s.err = s.loop()
	close(s.done)
}
227493
func (cl *Listener) Run(ctx context.Context) error {
	// Get our TLS config
	tlsConfig, err := cl.TLSConfig(ctx)
	if err != nil {
		cl.logger.Error("failed to get tls configuration when starting cluster listener", "error", err)
		return err
	}

	// The server supports all of the possible protos
	tlsConfig.NextProtos = []string{"h2", consts.RequestForwardingALPN, consts.PerfStandbyALPN, consts.PerformanceReplicationALPN, consts.DRReplicationALPN}

	for i, laddr := range cl.listenerAddrs {
		// closeCh is used to shutdown the spawned goroutines once this
		// function returns
		closeCh := make(chan struct{})

		if cl.logger.IsInfo() {
			cl.logger.Info("starting listener", "listener_address", laddr)
		}

		// Create a TCP listener. We do this separately and specifically
		// with TCP so that we can set deadlines.
		tcpLn, err := net.ListenTCP("tcp", laddr)
		if err != nil {
			cl.logger.Error("error starting listener", "error", err)
			continue
		}
		if laddr.String() != tcpLn.Addr().String() {
			// If we listened on port 0, record the port the OS gave us.
			cl.listenerAddrs[i] = tcpLn.Addr().(*net.TCPAddr)
		}

		// Wrap the listener with TLS
		tlsLn := tls.NewListener(tcpLn, tlsConfig)

		if cl.logger.IsInfo() {
			cl.logger.Info("serving cluster requests", "cluster_listen_address", tlsLn.Addr())
		}

		cl.shutdownWg.Add(1)
		// Start our listening loop
		go func(closeCh chan struct{}, tlsLn net.Listener) {
			defer func() {
				cl.shutdownWg.Done()
				tlsLn.Close()
				close(closeCh)
			}()

			for {
				if atomic.LoadUint32(cl.shutdown) > 0 {
					return
				}

				// Set the deadline for the accept call. If it passes we'll get
				// an error, causing us to check the condition at the top
				// again.
				tcpLn.SetDeadline(time.Now().Add(ListenerAcceptDeadline))

				// Accept the connection
				conn, err := tlsLn.Accept()
				if err != nil {
					if err, ok := err.(net.Error); ok && !err.Timeout() {
						cl.logger.Debug("non-timeout error accepting on cluster port", "error", err)
					}
					if conn != nil {
						conn.Close()
					}
					continue
				}
				if conn == nil {
					continue
				}

				// Type assert to TLS connection and handshake to populate the
				// connection state
				tlsConn := conn.(*tls.Conn)

				// Set a deadline for the handshake. This will cause clients
				// that don't successfully auth to be kicked out quickly.
				// Cluster connections should be reliable so being marginally
				// aggressive here is fine.
				err = tlsConn.SetDeadline(time.Now().Add(30 * time.Second))
				if err != nil {
					if cl.logger.IsDebug() {
						cl.logger.Debug("error setting deadline for cluster connection", "error", err)
					}
					tlsConn.Close()
					continue
				}

				err = tlsConn.Handshake()
				if err != nil {
					if cl.logger.IsDebug() {
						cl.logger.Debug("error handshaking cluster connection", "error", err)
					}
					tlsConn.Close()
					continue
				}

				// Now, set it back to unlimited
				err = tlsConn.SetDeadline(time.Time{})
				if err != nil {
					if cl.logger.IsDebug() {
						cl.logger.Debug("error setting deadline for cluster connection", "error", err)
					}
					tlsConn.Close()
					continue
				}

				cl.l.RLock()
				handler, ok := cl.handlers[tlsConn.ConnectionState().NegotiatedProtocol]
				cl.l.RUnlock()
				if !ok {
					cl.logger.Debug("unknown negotiated protocol on cluster port")
					tlsConn.Close()
					continue
				}

				if err := handler.Handoff(ctx, cl.shutdownWg, closeCh, tlsConn); err != nil {
					cl.logger.Error("error handling cluster connection", "error", err)
					continue
				}
			}
		}(closeCh, tlsLn)
	}

	return nil
}
1324405
func (p *UnitStatePublisher) Run(beatchan <-chan *unit.UnitStateHeartbeat, stop <-chan struct{}) {
	var period time.Duration
	if p.ttl > 10*time.Second {
		period = p.ttl * 4 / 5
	} else {
		period = p.ttl / 2
	}

	go func() {
		for {
			select {
			case <-stop:
				return
			case <-p.clock.After(period):
				p.cacheMutex.Lock()
				for name, us := range p.cache {
					go p.queueForPublish(name, us)
				}
				p.pruneCache()
				p.cacheMutex.Unlock()
			}
		}
	}()

	machID := p.mach.State().ID

	// Spawn goroutines to publish unit states. Each goroutine waits until
	// it sees an event arrive on toPublish, then attempts to grab the
	// relevant UnitState and publish it to the registry.
	for i := 0; i < numPublishers; i++ {
		go func() {
			for {
				select {
				case <-stop:
					return
				case name := <-p.toPublish:
					p.toPublishMutex.Lock()
					// Grab the latest state by that name
					us, ok := p.toPublishStates[name]
					if !ok {
						// If one doesn't exist, ignore.
						p.toPublishMutex.Unlock()
						continue
					}
					delete(p.toPublishStates, name)
					p.toPublishMutex.Unlock()
					p.publisher(name, us)

				}
			}
		}()
	}

	for {
		select {
		case <-stop:
			return
		case bt := <-beatchan:
			if bt.State != nil {
				bt.State.MachineID = machID
			}

			if p.updateCache(bt) {
				go p.queueForPublish(bt.Name, bt.State)
			}
		}
	}
}
1360324
func (mod *SocketBee) Run(cin chan bees.Event) {
	mod.eventChan = cin

	select {
	case <-mod.SigChan:
		return
	}
}
1016733
func (l *loop) run() {
	l.status = Running
	// Finalize the loop.
	defer l.finalizeTermination()
	// Create a loop wrapper containing the recovering control.
	loopWrapper := func() {
		defer func() {
			// Check for recovering.
			if reason := recover(); reason != nil {
				l.checkTermination(reason)
			}
		}()
		l.checkTermination(l.loopF(l))
	}
	// Now start running the loop wrappr.
	l.startedC <- struct{}{}
	for l.status == Running {
		loopWrapper()
	}
}
400677
func (sm *subscriptionManager) run() error {
	if sm.LoggingEnabled {
		log.Println("SubscriptionManager: Starting run.")
	}
	for {
		// NOTE: we check to see if its time to purge old buffers whenever
		// something happens or a period of inactivity has occurred.
		// An alternative would be to have another goroutine with a
		// select case time.After() but then you'd have concurrency issues
		// with access to the sm.SubEventBuffer and sm.bufferPriorityQueue objs
		// So instead of introducing mutexes we have this uglier manual time check calls
		select {
		case newClient := <-sm.clientSubscriptions:
			sm.handleNewClient(&newClient)
			sm.seeIfTimeToPurgeStaleCategories()
		case disconnected := <-sm.ClientTimeouts:
			sm.handleClientDisconnect(&disconnected)
			sm.seeIfTimeToPurgeStaleCategories()
		case event := <-sm.Events:
			sm.handleNewEvent(&event)
			sm.seeIfTimeToPurgeStaleCategories()
		case <-time.After(time.Duration(5) * time.Second):
			sm.seeIfTimeToPurgeStaleCategories()
		case _ = <-sm.Quit:
			if sm.LoggingEnabled {
				log.Println("SubscriptionManager: received quit signal, stopping.")
			}
			// break out of our infinite loop/select
			return nil
		}
	}
}
1107838
func (e *BackgroundTask) Run(m libkb.MetaContext) (err error) {
	defer m.Trace(e.Name(), func() error { return err })()

	// use a new background context with a saved cancel function
	var cancel func()
	m, cancel = m.BackgroundWithCancel()

	e.Lock()
	defer e.Unlock()

	e.shutdownFunc = cancel
	if e.shutdown {
		// Shutdown before started
		cancel()
		e.meta("early-shutdown")
		return nil
	}

	// start the loop and return
	go func() {
		err := e.loop(m)
		if err != nil {
			e.log(m, "loop error: %s", err)
		}
		cancel()
		e.meta("loop-exit")
	}()

	return nil
}
1160365
func (mod *RSSBee) Run(cin chan bees.Event) {
	mod.eventChan = cin

	time.Sleep(10 * time.Second)
	mod.pollFeed(mod.url, 5)
}
993348
func (n *Server) Run() {
	if err := server.RemoveUnixSocket(n.nccSocket); err != nil {
		log.Fatalf("Failed to remove socket: %v", err)
	}
	ln, err := net.Listen("unix", n.nccSocket)
	if err != nil {
		log.Fatalf("listen error: %v", err)
	}
	defer ln.Close()
	defer os.Remove(n.nccSocket)

	seesawNCC := rpc.NewServer()
	seesawNCC.Register(&SeesawNCC{})
	go server.RPCAccept(ln, seesawNCC)

	<-n.shutdown
}
1256188
func (c *CheckMonitor) run() {
	// Get the randomized initial pause time
	initialPauseTime := lib.RandomStagger(c.Interval)
	next := time.After(initialPauseTime)
	for {
		select {
		case <-next:
			c.check()
			next = time.After(c.Interval)
		case <-c.stopCh:
			return
		}
	}
}
1121314
func (c *NodeController) Run() {
	glog.Infof("starting node controller")

	stopCh := c.StopChannel()
	go c.runWatcher(stopCh)

	<-stopCh
	glog.Infof("shutting down node controller")
}
1048325
func run(c *Context, m, n []Middleware, h Handler) {
	switch {
	case len(m) > 0:
		if m[0] != nil {
			m[0](c, func(ctx *Context) { run(ctx, m[1:], n, h) })
		} else {
			run(c, m[1:], n, h)
		}
	case len(n) > 0:
		if n[0] != nil {
			n[0](c, func(ctx *Context) { run(ctx, nil, n[1:], h) })
		} else {
			run(c, nil, n[1:], h)
		}
	case h != nil:
		h(c)
	}
}
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)
					}
				}
			}
		}
	}
}
390151
func (m *Main) Run() (err error) {
	log.Printf("Running Main: %#v", m)
	var src pdk.Source
	if m.RegistryURL == "" {
		isrc := NewSource()
		isrc.Hosts = m.Hosts
		isrc.Topics = m.Topics
		isrc.Group = m.Group
		isrc.MaxMsgs = m.MaxRecords
		src = isrc
		if err := isrc.Open(); err != nil {
			return errors.Wrap(err, "opening kafka source")
		}
	} else {
		isrc := NewConfluentSource()
		isrc.Hosts = m.Hosts
		isrc.Topics = m.Topics
		isrc.Group = m.Group
		isrc.RegistryURL = m.RegistryURL
		isrc.MaxMsgs = m.MaxRecords
		src = isrc
		if err := isrc.Open(); err != nil {
			return errors.Wrap(err, "opening kafka source")
		}
	}

	if m.TranslatorDir == "" {
		m.TranslatorDir, err = ioutil.TempDir("", "pdk")
		if err != nil {
			return errors.Wrap(err, "creating temp directory")
		}
	}

	translateColumns := true
	parser := pdk.NewDefaultGenericParser()
	if len(m.SubjectPath) == 0 {
		parser.Subjecter = pdk.BlankSubjecter{}
		translateColumns = false
	} else {
		parser.EntitySubjecter = pdk.SubjectPath(m.SubjectPath)
	}

	mapper := pdk.NewCollapsingMapper()
	mapper.Translator, err = leveldb.NewTranslator(m.TranslatorDir)
	if err != nil {
		return errors.Wrap(err, "creating translator")
	}
	mapper.Framer = &m.Framer
	if translateColumns {
		log.Println("translating columns")
		mapper.ColTranslator, err = leveldb.NewFieldTranslator(m.TranslatorDir, "__columns")
		if err != nil {
			return errors.Wrap(err, "creating column translator")
		}
	} else {
		log.Println("not translating columns")
	}

	indexer, err := pdk.SetupPilosa(m.PilosaHosts, m.Index, nil, m.BatchSize)
	if err != nil {
		return errors.Wrap(err, "setting up Pilosa")
	}

	ingester := pdk.NewIngester(src, parser, mapper, indexer)
	if len(m.AllowedFields) > 0 {
		ingester.AllowedFields = make(map[string]bool)
		for _, fram := range m.AllowedFields {
			ingester.AllowedFields[fram] = true
		}
	}
	m.proxy = http.Server{
		Addr:    m.Proxy,
		Handler: pdk.NewPilosaForwarder(m.PilosaHosts[0], mapper.Translator, mapper.ColTranslator),
	}
	go func() {
		err := m.proxy.ListenAndServe()
		if err != nil {
			log.Printf("proxy closed: %v", err)
		}
	}()
	return errors.Wrap(ingester.Run(), "running ingester")
}