// BrokerFactory creates a new object of iface.Broker
// Currently only AMQP/S broker is supported
326541
func BrokerFactory(cnf *config.Config) (brokeriface.Broker, error) {
	if strings.HasPrefix(cnf.Broker, "amqp://") {
		return amqpbroker.New(cnf), nil
	}

	if strings.HasPrefix(cnf.Broker, "amqps://") {
		return amqpbroker.New(cnf), nil
	}

	if strings.HasPrefix(cnf.Broker, "redis://") {
		parts := strings.Split(cnf.Broker, "redis://")
		if len(parts) != 2 {
			return nil, fmt.Errorf(
				"Redis broker connection string should be in format redis://host:port, instead got %s",
				cnf.Broker,
			)
		}

		redisHost, redisPassword, redisDB, err := ParseRedisURL(cnf.Broker)
		if err != nil {
			return nil, err
		}
		return redisbroker.New(cnf, redisHost, redisPassword, "", redisDB), nil
	}

	if strings.HasPrefix(cnf.Broker, "redis+socket://") {
		redisSocket, redisPassword, redisDB, err := ParseRedisSocketURL(cnf.Broker)
		if err != nil {
			return nil, err
		}

		return redisbroker.New(cnf, "", redisPassword, redisSocket, redisDB), nil
	}

	if strings.HasPrefix(cnf.Broker, "eager") {
		return eagerbroker.New(), nil
	}

	if _, ok := os.LookupEnv("DISABLE_STRICT_SQS_CHECK"); ok {
		//disable SQS name check, so that users can use this with local simulated SQS
		//where sql broker url might not start with https://sqs

		//even when disabling strict SQS naming check, make sure its still a valid http URL
		if strings.HasPrefix(cnf.Broker, "https://") || strings.HasPrefix(cnf.Broker, "http://") {
			return sqsbroker.New(cnf), nil
		}
	} else {
		if strings.HasPrefix(cnf.Broker, "https://sqs") {
			return sqsbroker.New(cnf), nil
		}
	}

	if strings.HasPrefix(cnf.Broker, "gcppubsub://") {
		projectID, subscriptionName, err := ParseGCPPubSubURL(cnf.Broker)
		if err != nil {
			return nil, err
		}
		return gcppubsubbroker.New(cnf, projectID, subscriptionName)
	}

	return nil, fmt.Errorf("Factory failed with broker URL: %v", cnf.Broker)
}
1171427
func NewFactory(ctx context.Context) Factory {
	f := Factory{}
	f.typeRegistry = make(map[string]reflect.Type)
	f.typeRegistry["BarracudaNG"] = reflect.TypeOf(&BarracudaNG{}).Elem()
	f.typeRegistry["Iboss"] = reflect.TypeOf(&Iboss{}).Elem()
	f.typeRegistry["PaloAlto"] = reflect.TypeOf(&PaloAlto{}).Elem()
	f.typeRegistry["FortiGate"] = reflect.TypeOf(&FortiGate{}).Elem()
	f.typeRegistry["Checkpoint"] = reflect.TypeOf(&Checkpoint{}).Elem()
	f.typeRegistry["WatchGuard"] = reflect.TypeOf(&WatchGuard{}).Elem()
	f.typeRegistry["JSONRPC"] = reflect.TypeOf(&JSONRPC{}).Elem()
	f.typeRegistry["JuniperSRX"] = reflect.TypeOf(&JuniperSRX{}).Elem()
	return f
}
651936
func (f *ProviderFactory) CreateChannelProvider(config fab.EndpointConfig, opts ...options.Opt) (fab.ChannelProvider, error) {
	return chpvdr.New(config, opts...)
}
383658
func NewBrokerConf(clientID string) BrokerConf {
	return BrokerConf{
		ClientID:           clientID,
		DialTimeout:        10 * time.Second,
		DialRetryLimit:     10,
		DialRetryWait:      500 * time.Millisecond,
		AllowTopicCreation: false,
		LeaderRetryLimit:   10,
		LeaderRetryWait:    500 * time.Millisecond,
		RetryErrLimit:      10,
		RetryErrWait:       time.Millisecond * 500,
		ReadTimeout:        30 * time.Second,
		Logger:             &nullLogger{},
	}
}
883059
func New(cnf *config.Config) iface.Broker {
	b := &Broker{Broker: common.NewBroker(cnf)}
	if cnf.SQS != nil && cnf.SQS.Client != nil {
		// Use provided *SQS client
		b.service = cnf.SQS.Client
	} else {
		// Initialize a session that the SDK will use to load credentials from the shared credentials file, ~/.aws/credentials.
		// See details on: https://docs.aws.amazon.com/sdk-for-go/v1/developer-guide/configuring-sdk.html
		// Also, env AWS_REGION is also required
		b.sess = session.Must(session.NewSessionWithOptions(session.Options{
			SharedConfigState: session.SharedConfigEnable,
		}))
		b.service = awssqs.New(b.sess)
	}

	return b
}
382685
func (kz *Kazoo) brokerIDList() ([]int32, error) {
	brokers, err := kz.Brokers()
	if err != nil {
		return nil, err
	}

	result := make([]int32, 0, len(brokers))
	for id := range brokers {
		result = append(result, id)
	}

	// return sorted list to match the offical kafka sdks
	sort.Sort(int32Slice(result))

	return result, nil
}
638329
func (actor Actor) GetServiceBrokerByName(brokerName string) (ServiceBroker, Warnings, error) {
	serviceBrokers, warnings, err := actor.CloudControllerClient.GetServiceBrokers(ccv2.Filter{
		Type:     constant.NameFilter,
		Operator: constant.EqualOperator,
		Values:   []string{brokerName},
	})

	if err != nil {
		return ServiceBroker{}, Warnings(warnings), err
	}

	if len(serviceBrokers) == 0 {
		return ServiceBroker{}, Warnings(warnings), actionerror.ServiceBrokerNotFoundError{Name: brokerName}
	}

	return ServiceBroker(serviceBrokers[0]), Warnings(warnings), err
}
665928
func GetBasePreemptionResourceFactory() PreemptionResourceFactory {
	return func(availableResources *structs.ComparableResources, resourceNeeded *structs.ComparableResources) PreemptionResource {
		return &BasePreemptionResource{
			availableResources: availableResources,
			resourceNeeded:     resourceNeeded,
		}
	}
}
269785
func (r *router) getBroker(brokerAnnouncement *pb_discovery.Announcement) (*broker, error) {
	// We're going to be optimistic and guess that the broker is already active
	r.brokersLock.RLock()
	brk, ok := r.brokers[brokerAnnouncement.ID]
	r.brokersLock.RUnlock()
	if ok {
		return brk, nil
	}

	// If it doesn't we still have to lock
	r.brokersLock.Lock()
	defer r.brokersLock.Unlock()
	if _, ok := r.brokers[brokerAnnouncement.ID]; !ok {
		var err error

		brk := &broker{
			uplink: make(chan *pb_broker.UplinkMessage),
		}

		// Connect to the server
		brk.conn, err = brokerAnnouncement.Dial(r.Pool)
		if err != nil {
			return nil, err
		}

		// Set up the non-streaming client
		brk.client = pb_broker.NewBrokerClient(brk.conn)

		// Set up the streaming client
		config := brokerclient.DefaultClientConfig
		config.BackgroundContext = r.Component.Context
		cli := brokerclient.NewClient(config)
		cli.AddServer(brokerAnnouncement.ID, brk.conn)
		brk.association = cli.NewRouterStreams(r.Identity.ID, "")

		go func() {
			downlinkStream := brk.association.Downlink()
			refreshDownlinkStream := time.NewTicker(time.Second)
			for {
				select {
				case message := <-brk.uplink:
					brk.association.Uplink(message)
				case <-refreshDownlinkStream.C:
					downlinkStream = brk.association.Downlink()
				case message, ok := <-downlinkStream:
					if ok {
						go r.HandleDownlink(message)
					} else {
						r.Ctx.WithField("broker", brokerAnnouncement.ID).Error("Downlink Stream errored")
						downlinkStream = nil
					}
				}
			}
		}()

		r.brokers[brokerAnnouncement.ID] = brk
	}
	return r.brokers[brokerAnnouncement.ID], nil
}
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...)
	}
}
590129
func proxyFactory(protocol string, frontend ProxyAddress, backends ...ProxyAddress) (Proxy, error) {

	if len(backends) == 0 {
		return nil, errors.New("default proxy only requies one backend")
	}
	if len(backends) > 1 {
		return nil, errors.New("default proxy only supports one backend")
	}

	backendIP := net.ParseIP(backends[0].IP)
	if backendIP == nil {
		return nil, fmt.Errorf("not a valid IP format: %v", backendIP)
	}

	frontendIP := net.ParseIP(frontend.IP)
	if frontendIP == nil {
		return nil, fmt.Errorf("not a valid IP format: %v", frontendIP)
	}

	var frontendAddr, backendAddr net.Addr
	switch strings.Trim(strings.ToLower(protocol), " ") {
	case commons.TCP:
		frontendAddr = &net.TCPAddr{IP: frontendIP, Port: int(frontend.Port)}
		backendAddr = &net.TCPAddr{IP: backendIP, Port: int(backends[0].Port)}

	case commons.UDP:
		frontendAddr = &net.UDPAddr{IP: frontendIP, Port: int(frontend.Port)}
		backendAddr = &net.UDPAddr{IP: backendIP, Port: int(backends[0].Port)}

	default:
		return nil, fmt.Errorf("unsupported protocol %v", protocol)

	}

	proxy, err := proxy.NewProxy(frontendAddr, backendAddr)
	if err != nil {
		return nil, err
	}
	return &proxyWrapper{proxy}, nil
}
712945
func NewAMQPCeleryBrokerByConnAndChannel(conn *amqp.Connection, channel *amqp.Channel) *AMQPCeleryBroker {
	broker := &AMQPCeleryBroker{
		Channel:    channel,
		connection: conn,
		exchange:   NewAMQPExchange("default"),
		queue:      NewAMQPQueue("celery"),
		rate:       4,
	}
	if err := broker.CreateExchange(); err != nil {
		panic(err)
	}
	if err := broker.CreateQueue(); err != nil {
		panic(err)
	}
	if err := broker.Qos(broker.rate, 0, false); err != nil {
		panic(err)
	}
	if err := broker.StartConsumingChannel(); err != nil {
		panic(err)
	}
	return broker
}
1160135
func ClientOption(c ...option.ClientOption) broker.Option {
	return func(o *broker.Options) {
		if o.Context == nil {
			o.Context = context.Background()
		}
		o.Context = context.WithValue(o.Context, clientOptionKey{}, c)
	}
}
1014678
func newConfiguredProducer(logger *logrus.Entry, brokerString string, config *sarama.Config) (sarama.AsyncProducer, error) {
	brokerList := strings.Split(brokerString, ",")

	if len(brokerList) < 1 {
		logger.WithField("addrs", brokerString).Error("No brokers?")
		return nil, errors.New("No brokers in broker list")
	}

	logger.WithField("addrs", brokerList).Info("Connecting to Kafka")
	producer, err := sarama.NewAsyncProducer(brokerList, config)

	if err != nil {
		logger.Error("Error Connecting to Kafka. client error: ", err)
	}

	return producer, nil
}
1169831
func NewDefaultBroker() Broker {
	return &defaultBroker{
		routes:        make(map[URI]map[ID]Sender),
		subscriptions: make(map[ID]URI),
	}
}
818521
func NewFactory(buckets int) metrics.Factory {
	return adapters.WrapFactoryWithoutTags(
		&factory{
			factory: expvar.NewFactory(buckets),
		},
		adapters.Options{},
	)
}
1009537
func (kz *Kazoo) Brokers() (map[int32]string, error) {
	root := fmt.Sprintf("%s/brokers/ids", kz.conf.Chroot)
	children, _, err := kz.conn.Children(root)
	if err != nil {
		return nil, err
	}

	type brokerEntry struct {
		Host string `json:"host"`
		Port int    `json:"port"`
	}

	result := make(map[int32]string)
	for _, child := range children {
		brokerID, err := strconv.ParseInt(child, 10, 32)
		if err != nil {
			return nil, err
		}

		value, _, err := kz.conn.Get(path.Join(root, child))
		if err != nil {
			return nil, err
		}

		var brokerNode brokerEntry
		if err := json.Unmarshal(value, &brokerNode); err != nil {
			return nil, err
		}

		result[int32(brokerID)] = fmt.Sprintf("%s:%d", brokerNode.Host, brokerNode.Port)
	}

	return result, nil
}