// NewDispatcher builds a new Dispatcher using the specified Config. At
// minimum, a service name must be specified.
//
// Invalid configurations or errors in constructing the Dispatcher will cause
// panics.
545278
func NewDispatcher(cfg Config) *Dispatcher {
	if cfg.Name == "" {
		panic("yarpc.NewDispatcher expects a service name")
	}
	if err := internal.ValidateServiceName(cfg.Name); err != nil {
		panic("yarpc.NewDispatcher expects a valid service name: " + err.Error())
	}

	logger := cfg.Logging.logger(cfg.Name)
	extractor := cfg.Logging.extractor()

	meter, stopMeter := cfg.Metrics.scope(cfg.Name, logger)
	cfg = addObservingMiddleware(cfg, meter, logger, extractor)

	return &Dispatcher{
		name:              cfg.Name,
		table:             middleware.ApplyRouteTable(NewMapRouter(cfg.Name), cfg.RouterMiddleware),
		inbounds:          cfg.Inbounds,
		outbounds:         convertOutbounds(cfg.Outbounds, cfg.OutboundMiddleware),
		transports:        collectTransports(cfg.Inbounds, cfg.Outbounds),
		inboundMiddleware: cfg.InboundMiddleware,
		log:               logger,
		meter:             meter,
		stopMeter:         stopMeter,
		once:              lifecycle.NewOnce(),
	}
}
857527
func blobDispatcher(ctx *Context, r *http.Request) http.Handler {
	dgst, err := getDigest(ctx)
	if err != nil {

		if err == errDigestNotAvailable {
			return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
				ctx.Errors = append(ctx.Errors, v2.ErrorCodeDigestInvalid.WithDetail(err))
			})
		}

		return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
			ctx.Errors = append(ctx.Errors, v2.ErrorCodeDigestInvalid.WithDetail(err))
		})
	}

	blobHandler := &blobHandler{
		Context: ctx,
		Digest:  dgst,
	}

	mhandler := handlers.MethodHandler{
		"GET":  http.HandlerFunc(blobHandler.GetBlob),
		"HEAD": http.HandlerFunc(blobHandler.GetBlob),
	}

	if !ctx.readOnly {
		mhandler["DELETE"] = http.HandlerFunc(blobHandler.DeleteBlob)
	}

	return mhandler
}
202237
func (m *mapperDef) newMapper(c context.Context, j *mapper.Job, shardIdx int) (mapper.Mapper, error) {
	cfg := &api.JobConfig{}
	if err := proto.Unmarshal(j.Config.Params, cfg); err != nil {
		return nil, errors.Annotate(err, "failed to unmarshal JobConfig").Err()
	}
	return func(c context.Context, keys []*datastore.Key) error {
		return m.Func(c, j.ID, cfg, keys)
	}, nil
}
248888
func New(dispatcher eventservice.Dispatcher, opts ...options.Opt) *Client {
	params := defaultParams()
	options.Apply(params, opts)

	return &Client{
		Service:         eventservice.New(dispatcher, opts...),
		params:          *params,
		connectionState: int32(Disconnected),
	}
}
1342587
func (n *NeutrinoNotifier) notificationDispatcher() {
	defer n.wg.Done()
out:
	for {
		select {
		case cancelMsg := <-n.notificationCancels:
			switch msg := cancelMsg.(type) {
			case *epochCancel:
				chainntnfs.Log.Infof("Cancelling epoch "+
					"notification, epoch_id=%v", msg.epochID)

				// First, we'll lookup the original
				// registration in order to stop the active
				// queue goroutine.
				reg := n.blockEpochClients[msg.epochID]
				reg.epochQueue.Stop()

				// Next, close the cancel channel for this
				// specific client, and wait for the client to
				// exit.
				close(n.blockEpochClients[msg.epochID].cancelChan)
				n.blockEpochClients[msg.epochID].wg.Wait()

				// Once the client has exited, we can then
				// safely close the channel used to send epoch
				// notifications, in order to notify any
				// listeners that the intent has been
				// cancelled.
				close(n.blockEpochClients[msg.epochID].epochChan)
				delete(n.blockEpochClients, msg.epochID)
			}

		case registerMsg := <-n.notificationRegistry:
			switch msg := registerMsg.(type) {
			case *chainntnfs.HistoricalConfDispatch:
				// We'll start a historical rescan chain of the
				// chain asynchronously to prevent blocking
				// potentially long rescans.
				n.wg.Add(1)
				go func() {
					defer n.wg.Done()

					confDetails, err := n.historicalConfDetails(
						msg.ConfRequest,
						msg.StartHeight, msg.EndHeight,
					)
					if err != nil {
						chainntnfs.Log.Error(err)
						return
					}

					// If the historical dispatch finished
					// without error, we will invoke
					// UpdateConfDetails even if none were
					// found. This allows the notifier to
					// begin safely updating the height hint
					// cache at tip, since any pending
					// rescans have now completed.
					err = n.txNotifier.UpdateConfDetails(
						msg.ConfRequest, confDetails,
					)
					if err != nil {
						chainntnfs.Log.Error(err)
					}
				}()

			case *blockEpochRegistration:
				chainntnfs.Log.Infof("New block epoch subscription")

				n.blockEpochClients[msg.epochID] = msg

				// If the client did not provide their best
				// known block, then we'll immediately dispatch
				// a notification for the current tip.
				if msg.bestBlock == nil {
					n.notifyBlockEpochClient(
						msg, n.bestBlock.Height,
						n.bestBlock.Hash,
					)

					msg.errorChan <- nil
					continue
				}

				// Otherwise, we'll attempt to deliver the
				// backlog of notifications from their best
				// known block.
				n.bestBlockMtx.Lock()
				bestHeight := n.bestBlock.Height
				n.bestBlockMtx.Unlock()

				missedBlocks, err := chainntnfs.GetClientMissedBlocks(
					n.chainConn, msg.bestBlock, bestHeight,
					false,
				)
				if err != nil {
					msg.errorChan <- err
					continue
				}

				for _, block := range missedBlocks {
					n.notifyBlockEpochClient(
						msg, block.Height, block.Hash,
					)
				}

				msg.errorChan <- nil

			case *rescanFilterUpdate:
				err := n.chainView.Update(msg.updateOptions...)
				if err != nil {
					chainntnfs.Log.Errorf("Unable to "+
						"update rescan filter: %v", err)
				}
				msg.errChan <- err
			}

		case item := <-n.chainUpdates.ChanOut():
			update := item.(*filteredBlock)
			if update.connect {
				n.bestBlockMtx.Lock()
				// Since neutrino has no way of knowing what
				// height to rewind to in the case of a reorged
				// best known height, there is no point in
				// checking that the previous hash matches the
				// the hash from our best known height the way
				// the other notifiers do when they receive
				// a new connected block. Therefore, we just
				// compare the heights.
				if update.height != uint32(n.bestBlock.Height+1) {
					// Handle the case where the notifier
					// missed some blocks from its chain
					// backend
					chainntnfs.Log.Infof("Missed blocks, " +
						"attempting to catch up")

					_, missedBlocks, err :=
						chainntnfs.HandleMissedBlocks(
							n.chainConn,
							n.txNotifier,
							n.bestBlock,
							int32(update.height),
							false,
						)
					if err != nil {
						chainntnfs.Log.Error(err)
						n.bestBlockMtx.Unlock()
						continue
					}

					for _, block := range missedBlocks {
						filteredBlock, err :=
							n.getFilteredBlock(block)
						if err != nil {
							chainntnfs.Log.Error(err)
							n.bestBlockMtx.Unlock()
							continue out
						}
						err = n.handleBlockConnected(filteredBlock)
						if err != nil {
							chainntnfs.Log.Error(err)
							n.bestBlockMtx.Unlock()
							continue out
						}
					}

				}

				err := n.handleBlockConnected(update)
				if err != nil {
					chainntnfs.Log.Error(err)
				}

				n.bestBlockMtx.Unlock()
				continue
			}

			n.bestBlockMtx.Lock()
			if update.height != uint32(n.bestBlock.Height) {
				chainntnfs.Log.Infof("Missed disconnected " +
					"blocks, attempting to catch up")
			}
			newBestBlock, err := chainntnfs.RewindChain(
				n.chainConn, n.txNotifier, n.bestBlock,
				int32(update.height-1),
			)
			if err != nil {
				chainntnfs.Log.Errorf("Unable to rewind chain "+
					"from height %d to height %d: %v",
					n.bestBlock.Height, update.height-1, err)
			}

			// Set the bestHeight here in case a chain rewind
			// partially completed.
			n.bestBlock = newBestBlock
			n.bestBlockMtx.Unlock()

		case txUpdate := <-n.txUpdates.ChanOut():
			// A new relevant transaction notification has been
			// received from the backend. We'll attempt to process
			// it to determine if it fulfills any outstanding
			// confirmation and/or spend requests and dispatch
			// notifications for them.
			update := txUpdate.(*relevantTx)
			err := n.txNotifier.ProcessRelevantSpendTx(
				update.tx, uint32(update.details.Height),
			)
			if err != nil {
				chainntnfs.Log.Errorf("Unable to process "+
					"transaction %v: %v", update.tx.Hash(),
					err)
			}

		case err := <-n.rescanErr:
			chainntnfs.Log.Errorf("Error during rescan: %v", err)

		case <-n.quit:
			return

		}
	}
}
472613
func NewAllocator(lcDs, glDs datastore.DataStore) (*Allocator, error) {
	a := &Allocator{}

	// Load predefined subnet pools

	a.predefined = map[string][]*net.IPNet{
		localAddressSpace:  ipamutils.GetLocalScopeDefaultNetworks(),
		globalAddressSpace: ipamutils.GetGlobalScopeDefaultNetworks(),
	}

	// Initialize asIndices map
	a.predefinedStartIndices = make(map[string]int)

	// Initialize bitseq map
	a.addresses = make(map[SubnetKey]*bitseq.Handle)

	// Initialize address spaces
	a.addrSpaces = make(map[string]*addrSpace)
	for _, aspc := range []struct {
		as string
		ds datastore.DataStore
	}{
		{localAddressSpace, lcDs},
		{globalAddressSpace, glDs},
	} {
		a.initializeAddressSpace(aspc.as, aspc.ds)
	}

	return a, nil
}
582314
func NewGenerator(o *GeneratorOptions) Generator {
	// TODO(abg): Determine package name from `namespace go` directive.
	namespace := NewNamespace()
	return &generator{
		PackageName:    o.PackageName,
		ImportPath:     o.ImportPath,
		Namespace:      namespace,
		importer:       newImporter(namespace.Child()),
		mangler:        newMangler(),
		thriftImporter: o.Importer,
		fset:           token.NewFileSet(),
		noZap:          o.NoZap,
	}
}
838582
func NewHandler(h http.Handler) *GzipHandler {
	return &GzipHandler{
		wrap:    h,
		OnError: DefaultOnError,
	}
}
701565
func NewAdapter(registerHTTPHandler registerHTTPHandler, localtransp *syncbase.Registry) *Adapter {
	return &Adapter{registerHTTPHandler: registerHTTPHandler, base: localtransp}
}
616734
func (c *Configurator) NewDispatcherFromYAML(serviceName string, r io.Reader) (*yarpc.Dispatcher, error) {
	cfg, err := c.LoadConfigFromYAML(serviceName, r)
	if err != nil {
		return nil, err
	}
	return yarpc.NewDispatcher(cfg), nil
}
611348
func New(context context.Client, chConfig fab.ChannelCfg, discoveryService fab.DiscoveryService, connectionProvider api.ConnectionProvider, opts ...options.Opt) *Dispatcher {
	params := defaultParams(context, chConfig.ID())
	options.Apply(params, opts)

	dispatcher := &Dispatcher{
		Dispatcher:         esdispatcher.New(opts...),
		params:             *params,
		context:            context,
		chConfig:           chConfig,
		discoveryService:   discoveryService,
		connectionProvider: connectionProvider,
	}
	dispatcher.peerResolver = params.peerResolverProvider(dispatcher, context, chConfig.ID(), opts...)

	return dispatcher
}
904382
func NewRouter() *httprouter.Router {
	router := httprouter.New()
	router.NotFound = http.HandlerFunc(notFoundHandler)
	router.MethodNotAllowed = http.HandlerFunc(methodNotAllowedHandler)
	router.PanicHandler = panicHandler
	return router
}
762599
func InitDispatcher(d *tq.Dispatcher) {
	d.RegisterTask(&internal.EmailTask{}, SendEmail, "email", nil)
}
394271
func (ac *asyncCallbacksHandler) asyncCBDispatcher() {
	for {
		ac.mu.Lock()
		// Protect for spurious wakeups. We should get out of the
		// wait only if there is an element to pop from the list.
		for ac.head == nil {
			ac.cond.Wait()
		}
		cur := ac.head
		ac.head = cur.next
		if cur == ac.tail {
			ac.tail = nil
		}
		ac.mu.Unlock()

		// This signals that the dispatcher has been closed and all
		// previous callbacks have been dispatched.
		if cur.f == nil {
			return
		}
		// Invoke callback outside of handler's lock
		cur.f()
	}
}
878361
func NewPatchTransformer(
	slice []*resource.Resource, rf *resource.Factory) (transformers.Transformer, error) {
	if len(slice) == 0 {
		return transformers.NewNoOpTransformer(), nil
	}
	return &patchTransformer{patches: slice, rf: rf}, nil
}
434363
func (dc *DispatcherClient) NewBatch() *DispatcherBatch {
	return &DispatcherBatch{
		c: dc,
		b: dc.c.NewBatch(),
	}
}
75346
func (p *asyncProducer) dispatcher() {
	handlers := make(map[string]chan<- *ProducerMessage)
	shuttingDown := false

	for msg := range p.input {
		if msg == nil {
			Logger.Println("Something tried to send a nil message, it was ignored.")
			continue
		}

		if msg.flags&shutdown != 0 {
			shuttingDown = true
			p.inFlight.Done()
			continue
		} else if msg.retries == 0 {
			if shuttingDown {
				// we can't just call returnError here because that decrements the wait group,
				// which hasn't been incremented yet for this message, and shouldn't be
				pErr := &ProducerError{Msg: msg, Err: ErrShuttingDown}
				if p.conf.Producer.Return.Errors {
					p.errors <- pErr
				} else {
					Logger.Println(pErr)
				}
				continue
			}
			p.inFlight.Add(1)
		}

		version := 1
		if p.conf.Version.IsAtLeast(V0_11_0_0) {
			version = 2
		} else if msg.Headers != nil {
			p.returnError(msg, ConfigurationError("Producing headers requires Kafka at least v0.11"))
			continue
		}
		if msg.byteSize(version) > p.conf.Producer.MaxMessageBytes {
			p.returnError(msg, ErrMessageSizeTooLarge)
			continue
		}

		handler := handlers[msg.Topic]
		if handler == nil {
			handler = p.newTopicProducer(msg.Topic)
			handlers[msg.Topic] = handler
		}

		handler <- msg
	}

	for _, handler := range handlers {
		close(handler)
	}
}