// NewCeleryWorker returns new celery worker
545134
func NewCeleryWorker(broker CeleryBroker, backend CeleryBackend, numWorkers int) *CeleryWorker {
	return &CeleryWorker{
		broker:          broker,
		backend:         backend,
		numWorkers:      numWorkers,
		registeredTasks: map[string]interface{}{},
		rateLimitPeriod: 100 * time.Millisecond,
	}
}
1053086
func newWorker(a agent.Agent, apiCaller base.APICaller) (worker.Worker, error) {
	cfg := a.CurrentConfig()

	// Grab the tag and ensure that it's for a machine.
	tag, ok := cfg.Tag().(names.MachineTag)
	if !ok {
		return nil, errors.New("this manifold may only be used inside a machine agent")
	}

	// Get the machine agent's jobs.
	apiSt, err := apiagent.NewState(apiCaller)
	if err != nil {
		return nil, errors.Trace(err)
	}
	entity, err := apiSt.Entity(tag)
	if err != nil {
		return nil, err
	}
	var isModelManager bool
	for _, job := range entity.Jobs() {
		if job == multiwatcher.JobManageModel {
			isModelManager = true
			break
		}
	}

	if !isModelManager {
		return nil, dependency.ErrMissing
	}

	return NewWorker(cfg)
}
1137706
func NewWorker(config Config) (worker.Worker, error) {
	if err := config.Validate(); err != nil {
		return nil, errors.Trace(err)
	}
	p := &provisioner{config: config}
	err := catacomb.Invoke(catacomb.Plan{
		Site: &p.catacomb,
		Work: p.loop,
	})
	return p, err
}
217415
func NewWorker(numGoroutines int, addrs ...string) *Worker {

	pool := disque.NewPool(disque.DialFunc(func(addr string) (redis.Conn, error) {
		return redis.Dial("tcp", addr)
	}), addrs...)

	pool.RefreshNodes()
	pool.RunRefreshLoop()

	return &Worker{
		pool:          pool,
		numGoroutines: numGoroutines,
		workchan:      make(chan *Task),
		handlers:      map[string]TaskHandler{},
		stopch:        make(chan bool),
		channels:      []string{},
	}
}
308304
func New(pool *redis.Pool, src, id string) *DefaultWorker {
	heartbeater := heartbeat.New(
		id,
		fmt.Sprintf("%s:%s", src, "ticks"),
		defaultHeartInterval, pool)

	return &DefaultWorker{
		pool:           pool,
		availableTasks: queue.NewByteQueue(pool, src),
		workingTasks:   queue.NewDurableQueue(pool, src, getWorkingQueueName(src, id)),

		lifecycle: NewLifecycle(pool),
		detector:  heartbeater.Detector(),
		heart:     heartbeater.Heart(),
		janitor:   nilJanitor{},

		state: closed,
	}
}
948188
func (config ManifoldConfig) newWorker(a agent.Agent, apiCaller base.APICaller) (worker.Worker, error) {

	// This bit should be encapsulated in another manifold
	// satisfying jujud/agent/engine.Flag, as described in
	// the implementation below. Shouldn't be a concern here.
	if ok, err := isModelManager(a, apiCaller); err != nil {
		return nil, errors.Trace(err)
	} else if !ok {
		// This depends on a job change triggering an agent
		// bounce, which does happen today, but is not ideal;
		// another reason to use a flag.
		return nil, dependency.ErrMissing
	}

	// Get the API facade.
	if config.NewFacade == nil {
		logger.Errorf("nil NewFacade not valid, uninstalling")
		return nil, dependency.ErrUninstall
	}
	facade, err := config.NewFacade(apiCaller)
	if err != nil {
		return nil, errors.Trace(err)
	}

	// Start the worker.
	if config.NewWorker == nil {
		logger.Errorf("nil NewWorker not valid, uninstalling")
		return nil, dependency.ErrUninstall
	}
	worker, err := config.NewWorker(Config{
		Facade:   facade,
		Clock:    config.Clock,
		Interval: config.Interval,
	})
	if err != nil {
		return nil, errors.Trace(err)
	}
	return worker, nil
}
1331234
func NewMinUnitsWorker(st *state.State) worker.Worker {
	mu := &MinUnitsWorker{st: st}
	return legacy.NewStringsWorker(mu)
}
278638
func newWorkerNode() (*VSphere, error) {
	var err error
	vs := VSphere{}
	vs.hostName, err = os.Hostname()
	if err != nil {
		klog.Errorf("Failed to get hostname. err: %+v", err)
		return nil, err
	}
	vs.vmUUID, err = GetVMUUID()
	if err != nil {
		klog.Errorf("Failed to get uuid. err: %+v", err)
		return nil, err
	}
	return &vs, nil
}
690906
func NewWorker(id int, cl *trace.Client, logger *logrus.Logger, stats *statsd.Client) *Worker {
	return &Worker{
		id:               id,
		PacketChan:       make(chan samplers.UDPMetric, 32),
		ImportChan:       make(chan []samplers.JSONMetric, 32),
		ImportMetricChan: make(chan []*metricpb.Metric, 32),
		QuitChan:         make(chan struct{}),
		processed:        0,
		imported:         0,
		mutex:            &sync.Mutex{},
		traceClient:      cl,
		logger:           logger,
		wm:               NewWorkerMetrics(),
		stats:            stats,
	}
}
569025
func NewWorker(config Config) (worker.Worker, error) {
	if err := config.Validate(); err != nil {
		return nil, errors.Trace(err)
	}
	w := &forwarder{
		config: config,
	}
	unsubscribe, err := w.config.Hub.Subscribe(w.config.Topic, w.handleRequest)
	if err != nil {
		return nil, errors.Annotatef(err, "subscribing to %q", w.config.Topic)
	}
	w.unsubscribe = unsubscribe
	if err := catacomb.Invoke(catacomb.Plan{
		Site: &w.catacomb,
		Work: w.loop,
	}); err != nil {
		unsubscribe()
		return nil, errors.Trace(err)
	}
	return w, nil
}
972342
func New(tp TransactionPruner, interval time.Duration, clock clock.Clock) worker.Worker {
	return jworker.NewSimpleWorker(func(stopCh <-chan struct{}) error {
		for {
			select {
			case <-clock.After(interval):
				err := tp.MaybePruneTransactions()
				if err != nil {
					return errors.Annotate(err, "pruning failed, txnpruner stopping")
				}
			case <-stopCh:
				return nil
			}
		}
	})
}
285446
func newBlockRetrievalWorker(bg blockGetter, q *blockRetrievalQueue,
	workCh channels.Channel) *blockRetrievalWorker {
	brw := &blockRetrievalWorker{
		blockGetter: bg,
		stopCh:      make(chan struct{}),
		doneCh:      make(chan struct{}),
		queue:       q,
		workCh:      workCh,
	}
	go brw.run()
	return brw
}
327621
func NewWorker(config Config) (worker.Worker, error) {
	if err := config.Validate(); err != nil {
		return nil, errors.Trace(err)
	}

	rateLimitConfig, err := getRateLimitConfig(config.AgentConfig)
	if err != nil {
		return nil, errors.Annotate(err, "getting rate limit config")
	}

	logSinkConfig, err := getLogSinkConfig(config.AgentConfig)
	if err != nil {
		return nil, errors.Annotate(err, "getting log sink config")
	}

	controllerConfig, err := config.StatePool.SystemState().ControllerConfig()
	if err != nil {
		return nil, errors.Annotate(err, "cannot fetch the controller config")
	}

	observerFactory, err := newObserverFn(
		config.AgentConfig,
		controllerConfig,
		config.Clock,
		config.Hub,
		config.MetricsCollector,
	)
	if err != nil {
		return nil, errors.Annotate(err, "cannot create RPC observer factory")
	}

	serverConfig := apiserver.ServerConfig{
		StatePool:                     config.StatePool,
		Controller:                    config.Controller,
		Clock:                         config.Clock,
		Tag:                           config.AgentConfig.Tag(),
		DataDir:                       config.AgentConfig.DataDir(),
		LogDir:                        config.AgentConfig.LogDir(),
		Hub:                           config.Hub,
		Presence:                      config.Presence,
		Mux:                           config.Mux,
		Authenticator:                 config.Authenticator,
		RestoreStatus:                 config.RestoreStatus,
		UpgradeComplete:               config.UpgradeComplete,
		PublicDNSName:                 controllerConfig.AutocertDNSName(),
		AllowModelAccess:              controllerConfig.AllowModelAccess(),
		NewObserver:                   observerFactory,
		RegisterIntrospectionHandlers: config.RegisterIntrospectionHTTPHandlers,
		MetricsCollector:              config.MetricsCollector,
		RateLimitConfig:               rateLimitConfig,
		LogSinkConfig:                 &logSinkConfig,
		GetAuditConfig:                config.GetAuditConfig,
		LeaseManager:                  config.LeaseManager,
	}
	return config.NewServer(serverConfig)
}
459448
func newWorker(errorHandler func(error) error, closing chan bool, done chan bool) *worker {
	w := &worker{
		lock:         &sync.Mutex{},
		sink:         NewHTTPSink(),
		errorHandler: errorHandler,
		closing:      closing,
		done:         done,
	}

	return w
}
1062932
func NewWorker() (w *Worker) {
	astilog.Info("astiworker: starting worker...")
	w = &Worker{wg: &sync.WaitGroup{}}
	w.ctx, w.cancel = context.WithCancel(context.Background())
	w.wg.Add(1)
	return
}
79515
func s_worker_new(identity string) worker_t {
	return worker_t{
		identity:  identity,
		id_string: identity,
		expire:    time.Now().Add(HEARTBEAT_INTERVAL * HEARTBEAT_LIVENESS),
	}
}
640326
func NewWorker(config Config) (*Worker, error) {
	if err := config.Validate(); err != nil {
		return nil, errors.Trace(err)
	}

	w := &Worker{
		config: config,
		url:    make(chan string),
		status: "starting",
	}
	var err error
	var listener listener
	if w.config.ControllerAPIPort == 0 {
		listener, err = w.newSimpleListener()
	} else {
		listener, err = w.newDualPortListener()
	}
	if err != nil {
		return nil, errors.Trace(err)
	}
	w.holdable = newHeldListener(listener, config.Clock)

	if err := catacomb.Invoke(catacomb.Plan{
		Site: &w.catacomb,
		Work: w.loop,
	}); err != nil {
		listener.Close()
		return nil, errors.Trace(err)
	}
	return w, nil
}