// ObserverFactoryMultiplexer returns an ObserverFactory which will
// return a Multiplexer of all the observers instantiated from the
// factories passed in.
433039
func ObserverFactoryMultiplexer(factories ...ObserverFactory) ObserverFactory {
	return func() Observer {
		observers := make([]Observer, 0, len(factories))
		for _, newObserver := range factories {
			observers = append(observers, newObserver())
		}
		return NewMultiplexer(observers...)
	}
}
506912
func SimpleMuxer(fx func(Reactor, interface{})) SignalMuxHandler {
	return func(r Reactor, err error, data interface{}) {
		if err != nil {
			r.ReplyError(err)
			return
		}

		fx(r, data)
	}
}
749874
func NewAPIServiceInformer(client internalclientset.Interface, resyncPeriod time.Duration, indexers cache.Indexers) cache.SharedIndexInformer {
	return NewFilteredAPIServiceInformer(client, resyncPeriod, indexers, nil)
}
481886
func (d *DisallowFlunder) SetInternalWardleInformerFactory(f informers.SharedInformerFactory) {
	d.lister = f.Wardle().V1alpha1().Fischers().Lister()
	d.SetReadyFunc(f.Wardle().V1alpha1().Fischers().Informer().HasSynced)
}
1116505
func InterfaceMultiaddrs() ([]ma.Multiaddr, error) {
	addrs, err := net.InterfaceAddrs()
	if err != nil {
		return nil, err
	}

	maddrs := make([]ma.Multiaddr, len(addrs))
	for i, a := range addrs {
		maddrs[i], err = FromNetAddr(a)
		if err != nil {
			return nil, err
		}
	}
	return maddrs, nil
}
1054097
func NewMultiplexTransport(
	nodeInfo NodeInfo,
	nodeKey NodeKey,
	mConfig conn.MConnConfig,
) *MultiplexTransport {
	return &MultiplexTransport{
		acceptc:          make(chan accept),
		closec:           make(chan struct{}),
		dialTimeout:      defaultDialTimeout,
		filterTimeout:    defaultFilterTimeout,
		handshakeTimeout: defaultHandshakeTimeout,
		mConfig:          mConfig,
		nodeInfo:         nodeInfo,
		nodeKey:          nodeKey,
		conns:            NewConnSet(),
		resolver:         net.DefaultResolver,
	}
}
888613
func NewMultipartBuilder() MultipartBuilder {
	builder := MultipartBuilder{}
	var b bytes.Buffer
	builder.Buffer = &b
	builder.Writer = multipart.NewWriter(&b)
	return builder
}
1146785
func NewMultilineAdapter(route *router.Route) (a router.LogAdapter, err error) {
	enableByDefault := true
	enableStr := os.Getenv("MULTILINE_ENABLE_DEFAULT")
	if enableStr != "" {
		var err error
		enableByDefault, err = strconv.ParseBool(enableStr)
		if err != nil {
			return nil, errors.New("multiline: invalid value for MULTILINE_ENABLE_DEFAULT (must be true|false): " + enableStr)
		}
	}

	pattern := os.Getenv("MULTILINE_PATTERN")
	if pattern == "" {
		pattern = `^\s`
	}

	separator := os.Getenv("MULTILINE_SEPARATOR")
	if separator == "" {
		separator = "\n"
	}
	patternRegexp, err := regexp.Compile(pattern)
	if err != nil {
		return nil, errors.New("multiline: invalid value for MULTILINE_PATTERN (must be regexp): " + pattern)
	}

	matchType := os.Getenv("MULTILINE_MATCH")
	if matchType == "" {
		matchType = matchNonFirst
	}
	matchType = strings.ToLower(matchType)
	matchFirstLine := false
	negateMatch := false
	switch matchType {
	case matchFirst:
		matchFirstLine = true
		negateMatch = false
	case matchLast:
		matchFirstLine = false
		negateMatch = false
	case matchNonFirst:
		matchFirstLine = true
		negateMatch = true
	case matchNonLast:
		matchFirstLine = false
		negateMatch = true
	default:
		return nil, errors.New("multiline: invalid value for MULTILINE_MATCH (must be one of first|last|nonfirst|nonlast): " + matchType)
	}

	flushAfter := 500 * time.Millisecond
	flushAfterStr := os.Getenv("MULTILINE_FLUSH_AFTER")
	if flushAfterStr != "" {
		timeoutMS, err := strconv.Atoi(flushAfterStr)
		if err != nil {
			return nil, errors.New("multiline: invalid value for multiline_timeout (must be number): " + flushAfterStr)
		}
		flushAfter = time.Duration(timeoutMS) * time.Millisecond
	}

	parts := strings.SplitN(route.Adapter, "+", 2)
	if len(parts) != 2 {
		return nil, errors.New("multiline: adapter must have a sub-adapter, eg: multiline+raw+tcp")
	}

	originalAdapter := route.Adapter
	route.Adapter = parts[1]
	factory, found := router.AdapterFactories.Lookup(route.AdapterType())
	if !found {
		return nil, errors.New("bad adapter: " + originalAdapter)
	}
	subAdapter, err := factory(route)
	if err != nil {
		return nil, err
	}
	route.Adapter = originalAdapter

	out := make(chan *router.Message)
	checkInterval := flushAfter / 2

	return &Adapter{
		out:             out,
		subAdapter:      subAdapter,
		enableByDefault: enableByDefault,
		pattern:         patternRegexp,
		separator:       separator,
		matchFirstLine:  matchFirstLine,
		negateMatch:     negateMatch,
		flushAfter:      flushAfter,
		checkInterval:   checkInterval,
		buffers:         make(map[string]*router.Message),
		nextCheck:       time.After(checkInterval),
	}, nil
}
1011418
func (q *Query) Observer(observer QueryObserver) *Query {
	q.observer = observer
	return q
}
107458
func NewMultiSplitDiffWorker(wr *wrangler.Wrangler, cell, keyspace, shard string, excludeTables []string, minHealthyTablets, parallelDiffsCount int, waitForFixedTimeRatherThanGtidSet bool, useConsistentSnapshot bool, tabletType topodatapb.TabletType) Worker {
	return &MultiSplitDiffWorker{
		StatusWorker:                      NewStatusWorker(),
		wr:                                wr,
		cell:                              cell,
		keyspace:                          keyspace,
		shard:                             shard,
		excludeTables:                     excludeTables,
		minHealthyTablets:                 minHealthyTablets,
		parallelDiffsCount:                parallelDiffsCount,
		cleaner:                           &wrangler.Cleaner{},
		useConsistentSnapshot:             useConsistentSnapshot,
		waitForFixedTimeRatherThanGtidSet: waitForFixedTimeRatherThanGtidSet,
		tabletType:                        tabletType,
	}
}
1347324
func NewMultiwatcher(all *storeManager) *Multiwatcher {
	// Note that we want to be clear about the defaults. So we set zero
	// values explicitly.
	//  used:    false means that the watcher has not been used yet
	//  revno:   0 means that *all* transactions prior to the first
	//           Next() call will be reflected in the deltas.
	//  stopped: false means that the watcher immediately starts off
	//           handling changes.
	return &Multiwatcher{
		all:     all,
		used:    false,
		revno:   0,
		stopped: false,
	}
}
1215783
func WeakMultiplex(output SimpleInChannel, inputs ...SimpleOutChannel) {
	if len(inputs) == 0 {
		panic("channels: WeakMultiplex requires at least one input")
	}
	go multiplex(output, inputs, false)
}
404960
func (db *Database) GetMultipleFromView(designDoc string, view string,
	results interface{}, keys []string) error {
	var err error
	var url string
	type RequestBody struct {
		Keys []string `json:"keys"`
	}
	url, err = buildUrl(db.dbName, "_design", designDoc, "_view", view)
	if err != nil {
		return err
	}
	fmt.Errorf("url: " + url)
	var headers = make(map[string]string)
	reqBody := RequestBody{Keys: keys}
	requestBody, numBytes, err := encodeData(reqBody)
	if err != nil {
		return err
	}
	headers["Content-Type"] = "application/json"
	headers["Content-Length"] = strconv.Itoa(numBytes)
	if numBytes > 4000 {
		headers["Expect"] = "100-continue"
	}
	headers["Accept"] = "application/json"
	if resp, err :=
		db.connection.request("POST", url, requestBody,
			headers, db.auth); err == nil {
		defer resp.Body.Close()
		return parseBody(resp, &results)
	} else {
		return err
	}
}
246235
func (q *Queryx) Observer(observer gocql.QueryObserver) *Queryx {
	q.Query.Observer(observer)
	return q
}
770003
func NewMultiStream(
	listen net.Listener,
	dial func(string, time.Duration) (net.Conn, error),
	advertise net.Addr,
	lg *log.Logger,
	serviceCodes ...byte,
) (map[byte]raft.StreamLayer, error) {
	// sanity checks for listener and advertised address
	if advertise == nil {
		advertise = listen.Addr()
	}
	tcpAdvertise, ok := advertise.(*net.TCPAddr)
	if !ok {
		lg.Printf("NewMultiStream Warning: advertising non TCP address %s", advertise)
	} else if tcpAdvertise.IP.IsUnspecified() {
		lg.Printf("NewMultiStream Warning: advertising unspecified IP: %s for address %s", tcpAdvertise.IP, tcpAdvertise.String())
	}
	// set up router
	r := &router{listen, make(chan net.Conn, 16), make(chan closeReq, 16), make(map[byte]*serviceStreams), lg}
	// set up a channel of conns for each unique serviceCode
	for _, b := range serviceCodes {
		_, exists := r.chans[b]
		if exists {
			lg.Printf("Warning, serviceCode %d was present more than once in NewMultiStream -- allocating a single service")
			continue
		}
		r.chans[b] = &serviceStreams{
			r:      r,
			addr:   advertise,
			conns:  make(chan net.Conn, 16),
			closed: false,
			myCode: b,
			dial:   dial,
		}
	}
	// type hack, force to map[byte]raft.StreamLayer for return type
	retMap := make(map[byte]raft.StreamLayer)
	for k, v := range r.chans {
		retMap[k] = v
	}
	go r.serve()
	return retMap, nil
}
194924
func NewRecorderFactory(
	observerFactory rpc.ObserverFactory,
	recorder *auditlog.Recorder,
	captureArgs bool,
) rpc.RecorderFactory {
	return func() rpc.Recorder {
		return &combinedRecorder{
			observer:    observerFactory.RPCObserver(),
			recorder:    recorder,
			captureArgs: captureArgs,
		}
	}
}
941704
func (device *Device) ObserverFillElapsed() float64 {
	ret := float64(C.cairo_device_observer_fill_elapsed(device.Ptr))
	if err := device.status(); err != nil {
		panic(err)
	}
	return ret
}