// Run starts the export listener for the given tenant
779462
func (l *ImportListener) Run(cancel <-chan struct{}, conn client.Connection) {
	logger := plog.WithField("tenantid", l.tenantID)

	pth := path.Join("/net/export", l.tenantID)

	done := make(chan struct{})
	defer func() { close(done) }()
	for {

		// set a watcher to alert when the path is available
		ok, ev, err := conn.ExistsW(pth, done)
		if err != nil {
			logger.WithError(err).Error("Could not listen for exports on tenant")
			return
		}

		var ch []string
		if ok {
			// the path is available, so set the listener on the change in the
			// number of children
			ch, ev, err = conn.ChildrenW(pth, done)
			if err == client.ErrNoNode {
				logger.Debug("Import was suddenly deleted, retrying")
				close(done)
				done = make(chan struct{})

				continue
			} else if err != nil {
				logger.WithError(err).Error("Could not listen for exports on tenant")
				return
			}
		}

		// for each new application found, send the matches to the respective
		// term channels.
		for _, app := range ch {
			if _, ok := l.apps[app]; !ok {
				for _, t := range l.terms {
					if t.matcher.MatchString(app) {
						select {
						case t.send <- app:
						case <-cancel:
							return
						}
					}
				}
				l.apps[app] = struct{}{}
			}
		}

		// wait for something to happen
		select {
		case <-ev:
		case <-cancel:
			return
		}

		close(done)
		done = make(chan struct{})
	}
}
393355
func (s *Server) Run() {
	go s.updater()
	go s.notifier()
	go s.manager()

	<-s.quit
}
30484
func (c *MockSSHClient) Run(command string, stdout io.Writer, stderr io.Writer) error {
	if c.MockRun != nil {
		return c.MockRun(command, stdout, stderr)
	}
	return ErrNotImplemented
}
461003
func (cmd *GenerateConfigCommand) Run(_ context.Context) error {
	conf := server.NewConfig()
	ret, err := toml.Marshal(*conf)
	if err != nil {
		return errors.Wrap(err, "unmarshalling default config")
	}
	fmt.Fprintf(cmd.Stdout, "%s\n", ret)
	return nil
}
650559
func (e *Archive) Run(c *fi.Context) error {
	return fi.DefaultDeltaRunMethod(e, c)
}
939355
func (u *UIDMonitor) Run(ctx context.Context) error {

	if err := u.proc.config.IsComplete(); err != nil {
		return fmt.Errorf("uid: %s", err)
	}

	return u.Resync(ctx)
}
1305966
func (t *CopySchemaShardTask) Run(parameters map[string]string) ([]*automationpb.TaskContainer, string, error) {
	args := []string{"CopySchemaShard"}
	if tables := parameters["tables"]; tables != "" {
		args = append(args, "--tables="+tables)
	}
	if excludeTables := parameters["exclude_tables"]; excludeTables != "" {
		args = append(args, "--exclude_tables="+excludeTables)
	}
	args = append(args, parameters["source_keyspace_and_shard"], parameters["dest_keyspace_and_shard"])
	output, err := ExecuteVtctl(context.TODO(), parameters["vtctld_endpoint"], args)
	return nil, output, err
}
234943
func (c *DeploymentConfigController) Run(workers int, stopCh <-chan struct{}) {
	defer utilruntime.HandleCrash()
	defer c.queue.ShutDown()

	klog.Infof("Starting deploymentconfig controller")

	// Wait for the rc and dc stores to sync before starting any work in this controller.
	if !cache.WaitForCacheSync(stopCh, c.dcStoreSynced, c.rcListerSynced) {
		return
	}

	klog.Info("deploymentconfig controller caches are synced. Starting workers.")

	for i := 0; i < workers; i++ {
		go wait.Until(c.worker, time.Second, stopCh)
	}

	metrics.InitializeMetricsCollector(c.rcLister)

	<-stopCh

	klog.Infof("Shutting down deploymentconfig controller")
}
941062
func (t *Typescript) Run(logger func(format string, a ...interface{})) {
	c := t.Config
	if c.Tsconfig == nil {
		tsC := DefaultTsconfig()
		c.Tsconfig = &tsC
	}

	if c.Dir == "" {
		c.Tsconfig.CompilerOptions.OutDir = c.Dir
	}

	if c.Dir == "" {
		c.Dir = "./"
	}

	if !strings.Contains(c.Ignore, nodeModules) {
		c.Ignore += "," + nodeModules
	}

	if logger == nil {
		logger = NoOpLogger
	}

	t.log = logger

	t.start()
}
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)
			}
		}
	}
}
195650
func (d *Dispatcher) Run(ctx context.Context) error {
	ctx = log.WithModule(ctx, "dispatcher")
	log.G(ctx).Info("dispatcher starting")

	d.taskUpdatesLock.Lock()
	d.taskUpdates = make(map[string]*api.TaskStatus)
	d.taskUpdatesLock.Unlock()

	d.nodeUpdatesLock.Lock()
	d.nodeUpdates = make(map[string]nodeUpdate)
	d.nodeUpdatesLock.Unlock()

	d.mu.Lock()
	if d.isRunning() {
		d.mu.Unlock()
		return errors.New("dispatcher is already running")
	}
	if err := d.markNodesUnknown(ctx); err != nil {
		log.G(ctx).Errorf(`failed to move all nodes to "unknown" state: %v`, err)
	}
	configWatcher, cancel, err := store.ViewAndWatch(
		d.store,
		func(readTx store.ReadTx) error {
			clusters, err := store.FindClusters(readTx, store.ByName(store.DefaultClusterName))
			if err != nil {
				return err
			}
			if err == nil && len(clusters) == 1 {
				heartbeatPeriod, err := gogotypes.DurationFromProto(clusters[0].Spec.Dispatcher.HeartbeatPeriod)
				if err == nil && heartbeatPeriod > 0 {
					d.config.HeartbeatPeriod = heartbeatPeriod
				}
				if clusters[0].NetworkBootstrapKeys != nil {
					d.networkBootstrapKeys = clusters[0].NetworkBootstrapKeys
				}
				d.lastSeenRootCert = clusters[0].RootCA.CACert
			}
			return nil
		},
		api.EventUpdateCluster{},
	)
	if err != nil {
		d.mu.Unlock()
		return err
	}
	// set queue here to guarantee that Close will close it
	d.clusterUpdateQueue = watch.NewQueue()

	peerWatcher, peerCancel := d.cluster.SubscribePeers()
	defer peerCancel()
	d.lastSeenManagers = getWeightedPeers(d.cluster)

	defer cancel()
	d.ctx, d.cancel = context.WithCancel(ctx)
	ctx = d.ctx
	d.wg.Add(1)
	defer d.wg.Done()
	d.mu.Unlock()

	publishManagers := func(peers []*api.Peer) {
		var mgrs []*api.WeightedPeer
		for _, p := range peers {
			mgrs = append(mgrs, &api.WeightedPeer{
				Peer:   p,
				Weight: remotes.DefaultObservationWeight,
			})
		}
		d.mu.Lock()
		d.lastSeenManagers = mgrs
		d.mu.Unlock()
		d.clusterUpdateQueue.Publish(clusterUpdate{managerUpdate: &mgrs})
	}

	batchTimer := time.NewTimer(maxBatchInterval)
	defer batchTimer.Stop()

	for {
		select {
		case ev := <-peerWatcher:
			publishManagers(ev.([]*api.Peer))
		case <-d.processUpdatesTrigger:
			d.processUpdates(ctx)
			batchTimer.Stop()
			// drain the timer, if it has already expired
			select {
			case <-batchTimer.C:
			default:
			}
			batchTimer.Reset(maxBatchInterval)
		case <-batchTimer.C:
			d.processUpdates(ctx)
			// batch timer has already expired, so no need to drain
			batchTimer.Reset(maxBatchInterval)
		case v := <-configWatcher:
			cluster := v.(api.EventUpdateCluster)
			d.mu.Lock()
			if cluster.Cluster.Spec.Dispatcher.HeartbeatPeriod != nil {
				// ignore error, since Spec has passed validation before
				heartbeatPeriod, _ := gogotypes.DurationFromProto(cluster.Cluster.Spec.Dispatcher.HeartbeatPeriod)
				if heartbeatPeriod != d.config.HeartbeatPeriod {
					// only call d.nodes.updatePeriod when heartbeatPeriod changes
					d.config.HeartbeatPeriod = heartbeatPeriod
					d.nodes.updatePeriod(d.config.HeartbeatPeriod, d.config.HeartbeatEpsilon, d.config.GracePeriodMultiplier)
				}
			}
			d.lastSeenRootCert = cluster.Cluster.RootCA.CACert
			d.networkBootstrapKeys = cluster.Cluster.NetworkBootstrapKeys
			d.mu.Unlock()
			d.clusterUpdateQueue.Publish(clusterUpdate{
				bootstrapKeyUpdate: &cluster.Cluster.NetworkBootstrapKeys,
				rootCAUpdate:       &cluster.Cluster.RootCA.CACert,
			})
		case <-ctx.Done():
			return nil
		}
	}
}
444036
func (r *Rabbus) Run(ctx context.Context) error {
	notifyClose := r.NotifyClose(make(chan *amqp.Error))

	for {
		select {
		case m, ok := <-r.emit:
			if !ok {
				return errors.New("unexpected close of emit channel")
			}

			r.produce(m)

		case err := <-notifyClose:
			if err == nil {
				// "… on a graceful close, no error will be sent."
				return nil
			}

			r.handleAmqpClose(err)

			// We have reconnected, so we need a new NotifyClose again.
			notifyClose = r.NotifyClose(make(chan *amqp.Error))

		case <-ctx.Done():
			return ctx.Err()
		}
	}
}
791967
func (m *Main) Run() error {
	m.Logger.Info("running")

	// Read or generate the instance identifier from file.
	id, err := m.readID(filepath.Join(m.DataDir, "instance_id"))
	if err != nil {
		return err
	}
	m.Process.ID = id
	m.Process.DataDir = m.DataDir
	m.Process.Logger = m.Logger.New("component", "process", "id", id)

	// Start process.
	m.Logger.Info("starting process", "id", id)
	if err := m.Process.Start(); err != nil {
		m.Logger.Error("error starting process", "err", err)
		return err
	}

	// Add service to discoverd registry.
	m.Logger.Info("adding service", "name", m.ServiceName)
	if err = m.DiscoverdClient.AddService(m.ServiceName, nil); err != nil && !httphelper.IsObjectExistsError(err) {
		m.Logger.Error("error adding discoverd service", "err", err)
		return err
	}
	inst := &discoverd.Instance{
		Addr: ":" + m.Process.Port,
		Meta: map[string]string{"REDIS_ID": id},
	}

	// Register instance and retain heartbeater.
	m.Logger.Info("registering instance", "addr", inst.Addr, "meta", inst.Meta)
	hb, err := m.DiscoverdClient.RegisterInstance(m.ServiceName, inst)
	if err != nil {
		m.Logger.Error("error registering discoverd instance", "err", err)
		return err
	}
	m.hb = hb
	shutdown.BeforeExit(func() { hb.Close() })

	m.Logger.Info("opening port", "addr", m.Addr)

	// Open HTTP port.
	ln, err := net.Listen("tcp", m.Addr)
	if err != nil {
		m.Logger.Error("error opening port", "err", err)
		return err
	}
	m.ln = keepalive.Listener(ln)

	// Initialize and server handler.
	m.Logger.Info("serving http api")
	h := redis.NewHandler()
	h.Process = m.Process
	h.Heartbeater = m.hb
	h.Logger = m.Logger.New("component", "http")
	go func() { http.Serve(ln, h) }()

	return nil
}
1069527
func (c *controller) Run(stopCh <-chan struct{}) {
	if c.leaseClient == nil {
		klog.Infof("node lease controller has nil lease client, will not claim or renew leases")
		return
	}
	wait.Until(c.sync, c.renewInterval, stopCh)
}
1361897
func (p *AttachOptions) Run() error {
	pod, err := p.Client.Pods(p.Namespace).Get(p.PodName)
	if err != nil {
		return err
	}

	if pod.Status.Phase != api.PodRunning {
		return fmt.Errorf("pod %s is not running and cannot be attached to; current phase is %s", p.PodName, pod.Status.Phase)
	}

	var stdin io.Reader
	tty := p.TTY

	containerToAttach := p.GetContainer(pod)
	if tty && !containerToAttach.TTY {
		tty = false
		fmt.Fprintf(p.Err, "Unable to use a TTY - container %s doesn't allocate one\n", containerToAttach.Name)
	}

	// TODO: refactor with terminal helpers from the edit utility once that is merged
	if p.Stdin {
		stdin = p.In
		if tty {
			if file, ok := stdin.(*os.File); ok {
				inFd := file.Fd()
				if term.IsTerminal(inFd) {
					oldState, err := term.SetRawTerminal(inFd)
					if err != nil {
						glog.Fatal(err)
					}
					fmt.Fprintln(p.Out, "\nHit enter for command prompt")
					// this handles a clean exit, where the command finished
					defer term.RestoreTerminal(inFd, oldState)

					// SIGINT is handled by term.SetRawTerminal (it runs a goroutine that listens
					// for SIGINT and restores the terminal before exiting)

					// this handles SIGTERM
					sigChan := make(chan os.Signal, 1)
					signal.Notify(sigChan, syscall.SIGTERM)
					go func() {
						<-sigChan
						term.RestoreTerminal(inFd, oldState)
						os.Exit(0)
					}()
				} else {
					fmt.Fprintln(p.Err, "STDIN is not a terminal")
				}
			} else {
				tty = false
				fmt.Fprintln(p.Err, "Unable to use a TTY - input is not the right kind of file")
			}
		}
	}

	// TODO: consider abstracting into a client invocation or client helper
	req := p.Client.RESTClient.Post().
		Resource("pods").
		Name(pod.Name).
		Namespace(pod.Namespace).
		SubResource("attach")
	req.VersionedParams(&api.PodAttachOptions{
		Container: containerToAttach.Name,
		Stdin:     stdin != nil,
		Stdout:    p.Out != nil,
		Stderr:    p.Err != nil,
		TTY:       tty,
	}, api.Scheme)

	return p.Attach.Attach("POST", req.URL(), p.Config, stdin, p.Out, p.Err, tty)
}
801846
func Run() error {
	if service.Registry != nil {
		service.Registry.Register(service)
	}

	// Start a metrics server in the background
	startPrometheusServer()

	// Create and then server a gRPC server
	err := ServeGRPC()
	if service.Registry != nil {
		service.Registry.DeRegister(service)
	}
	return err
}
903420
func (c *CmdSimpleFSDebug) Run() error {
	cli, err := GetSimpleFSClient(c.G())
	if err != nil {
		return err
	}

	return cli.SimpleFSDumpDebuggingInfo(context.TODO())
}