// newConsumer returns a new consumer group using default configuration.
772952
func newConsumer(brokers []string, topic string, group string, offset int64, sync bool) (*ConsumerGroup, error) {

	config := cluster.NewConfig()
	config.Group.Return.Notifications = true
	config.Consumer.Return.Errors = true
	config.Consumer.MaxWaitTime = 50 * time.Millisecond
	config.Consumer.Offsets.Initial = offset
	config.Consumer.Offsets.Retention = 0 // indefinite retention

	logData := log.Data{"topic": topic, "group": group, "config": config}

	consumer, err := cluster.NewConsumer(brokers, group, []string{topic}, config)
	if err != nil {
		log.ErrorC("newConsumer failed", err, logData)
		return nil, err
	}

	var upstream chan Message
	if sync {
		// make the upstream channel buffered, so we can send-and-wait for upstreamDone
		upstream = make(chan Message, 1)
	} else {
		upstream = make(chan Message)
	}

	cg := &ConsumerGroup{
		consumer:     consumer,
		incoming:     upstream,
		closer:       make(chan struct{}),
		closed:       make(chan struct{}),
		errors:       make(chan error),
		topic:        topic,
		group:        group,
		sync:         sync,
		upstreamDone: make(chan bool, 1),
	}

	// listener goroutine - listen to consumer.Messages() and upstream them
	// if this blocks while upstreaming a message, we can shutdown consumer via the following goroutine
	go func() {
		logData := log.Data{"topic": topic, "group": group, "config": config}

		log.Info("Started kafka consumer listener", logData)
		defer close(cg.closed)
		for looping := true; looping; {
			select {
			case <-cg.closer:
				looping = false
			case msg := <-cg.consumer.Messages():
				cg.Incoming() <- SaramaMessage{msg, cg.consumer}
				if cg.sync {
					// wait for msg-processed or close-consumer triggers
					for loopingForSync := true; looping && loopingForSync; {
						select {
						case <-cg.upstreamDone:
							loopingForSync = false
						case <-cg.closer:
							// XXX if we read closer here, this means that the release/upstreamDone blocks unless it is buffered
							looping = false
						}
					}
				}
			}
		}
		cg.consumer.CommitOffsets()
		log.Info("Closed kafka consumer listener", logData)
	}()

	// control goroutine - allows us to close consumer even if blocked while upstreaming a message (above)
	go func() {
		logData := log.Data{"topic": topic, "group": group}

		hasBalanced := false // avoid CommitOffsets() being called before we have balanced (otherwise causes a panic)
		for looping := true; looping; {
			select {
			case <-cg.closer:
				log.Info("Closing kafka consumer controller", logData)
				<-cg.closed
				looping = false
			case err := <-cg.consumer.Errors():
				log.Error(err, nil)
				cg.Errors() <- err
			case <-time.After(tick):
				if hasBalanced {
					cg.consumer.CommitOffsets()
				}
			case n, more := <-cg.consumer.Notifications():
				if more {
					hasBalanced = true
					log.Trace("Rebalancing group", log.Data{"topic": cg.topic, "group": cg.group, "partitions": n.Current[cg.topic]})
				}
			}
		}
		log.Info("Closed kafka consumer controller", logData)
	}()

	return cg, nil
}
643999
func NewGroupMembersSelector(Group *GroupSelector, Users *UsersSelectorArg) *GroupMembersSelector {
	s := new(GroupMembersSelector)
	s.Group = Group
	s.Users = Users
	return s
}
340056
func (this *ZookeeperCoordinator) RegisterConsumer(Consumerid string, Groupid string, TopicCount TopicsToNumStreams) (err error) {
	backoffMultiplier := 1
	this.ensureZkPathsExist(Groupid)
	for i := 0; i <= this.config.MaxRequestRetries; i++ {
		err = this.tryRegisterConsumer(Consumerid, Groupid, TopicCount)
		if err == nil {
			return
		}
		Tracef(this, "Registering consumer %s in group %s failed after %d-th retry", Consumerid, Groupid, i)
		time.Sleep(this.config.RequestBackoff * time.Duration(backoffMultiplier))
		backoffMultiplier++
	}
	return
}
769035
func (e *Login) SubConsumers() []libkb.UIConsumer {
	return []libkb.UIConsumer{
		&LoginProvisionedDevice{},
		&loginLoadUser{},
		&loginProvision{},
		&AccountReset{},
	}
}
430838
func (d *API) RegisterConsumer(mediaType string, handler runtime.Consumer) {
	if d.consumers == nil {
		d.consumers = make(map[string]runtime.Consumer, 10)
	}
	d.consumers[strings.ToLower(mediaType)] = handler
}
619312
func NewOrdererConfig(ordererGroup *cb.ConfigGroup, mspConfig *MSPConfigHandler) (*OrdererConfig, error) {
	oc := &OrdererConfig{
		protos: &OrdererProtos{},
		orgs:   make(map[string]Org),
	}

	if err := DeserializeProtoValuesFromGroup(ordererGroup, oc.protos); err != nil {
		return nil, errors.Wrap(err, "failed to deserialize values")
	}

	if err := oc.Validate(); err != nil {
		return nil, err
	}

	for orgName, orgGroup := range ordererGroup.Groups {
		var err error
		if oc.orgs[orgName], err = NewOrganizationConfig(orgName, orgGroup, mspConfig); err != nil {
			return nil, err
		}
	}
	return oc, nil
}
409889
func NewConsumerCodec(f func(io.ReadWriteCloser) rpc.ClientCodec) *rpc.Client {
	return rpc.NewClientWithCodec(f(rwCloser{os.Stdin, os.Stdout}))
}
98745
func (cg *Consumergroup) Create() error {
	return cg.kz.mkdirRecursive(fmt.Sprintf("%s/consumers/%s", cg.kz.conf.Chroot, cg.Name))
}
655772
func NewConsumer(s *SQSService, handler MessageHandlerFunc) *Consumer {
	return &Consumer{
		s:                              s,
		handler:                        handler,
		delayAfterReceiveError:         defaultDelayAfterReceiveError,
		WaitSeconds:                    defaultReceiveMessageWaitSeconds,
		ReceiveVisibilityTimoutSeconds: defaultReceiveVisibilityTimeoutSeconds,
		Logger: NoopLogger,

		ExtendVisibilityTimeoutBySeconds: defaultExtendVisibilityBySeconds,
		ExtendVisibilityTimeoutEvery:     defaultExtendVisibilityEvery,
		DeleteMessageAccumulatorTimeout:  defaultDeleteAccumulatorTimeout,
		DeleteMessageDrainTimeout:        defaultDeleteMessageDrainTimeout,
	}
}
114855
func (this *ZookeeperCoordinator) DeregisterConsumer(Consumerid string, Groupid string) (err error) {
	path := fmt.Sprintf("%s/%s", newZKGroupDirs(this.config.Root, Groupid).ConsumerRegistryDir, Consumerid)
	Debugf(this, "Trying to deregister consumer at path: %s", path)
	backoffMultiplier := 1
	for i := 0; i <= this.config.MaxRequestRetries; i++ {
		err = this.deleteNode(path)
		if err == nil {
			return
		}
		Tracef(this, "Deregistering consumer %s in group %s failed after %d-th retry", Consumerid, Groupid, i)
		time.Sleep(this.config.RequestBackoff * time.Duration(backoffMultiplier))
		backoffMultiplier++
	}
	return
}
1334253
func (s *ListStreamConsumersOutput) SetConsumers(v []*Consumer) *ListStreamConsumersOutput {
	s.Consumers = v
	return s
}
987007
func NewConsortiumsGroup(conf map[string]*genesisconfig.Consortium) (*cb.ConfigGroup, error) {
	consortiumsGroup := cb.NewConfigGroup()
	// This policy is not referenced anywhere, it is only used as part of the implicit meta policy rule at the channel level, so this setting
	// effectively degrades control of the ordering system channel to the ordering admins
	addPolicy(consortiumsGroup, policies.SignaturePolicy(channelconfig.AdminsPolicyKey, cauthdsl.AcceptAllPolicy), ordererAdminsPolicyName)

	for consortiumName, consortium := range conf {
		var err error
		consortiumsGroup.Groups[consortiumName], err = NewConsortiumGroup(consortium)
		if err != nil {
			return nil, errors.Wrapf(err, "failed to create consortium %s", consortiumName)
		}
	}

	consortiumsGroup.ModPolicy = ordererAdminsPolicyName
	return consortiumsGroup, nil
}
132225
func (cfg *Config) NewDecoder(r io.Reader) *Decoder {
	return &Decoder{r, cfg}
}
29753
func NewGroupMemberSelector(Group *GroupSelector, User *UserSelectorArg) *GroupMemberSelector {
	s := new(GroupMemberSelector)
	s.Group = Group
	s.User = User
	return s
}
1353371
func (mux *Multiplexer) genericConsumer() {
	mux.Debug("Generic Consumer started")
	for {
		select {
		case <-mux.Consumer.GetCloseChannel():
			mux.Debug("Closing Consumer")
			return
		case msg := <-mux.Consumer.Config.RecvMessageChan:
			// 'hash' partitioner messages will be marked
			mux.propagateMessage(msg)
		case err := <-mux.Consumer.Config.RecvErrorChan:
			mux.Error("Received partitionConsumer error ", err)
		}
	}
}
764698
func NewConsumer(client Client, config *ConsumerConfig, strategy Strategy) Consumer {
	var metrics ConsumerMetrics = noOpConsumerMetrics
	if config.EnableMetrics {
		metrics = NewKafkaConsumerMetrics(config.Group, config.ConsumerID)
	}

	return &KafkaConsumer{
		config:                   config,
		client:                   client,
		strategy:                 strategy,
		metrics:                  metrics,
		partitionConsumers:       make(map[string]map[int32]PartitionConsumer),
		partitionConsumerFactory: NewPartitionConsumer,
		stopped:                  make(chan struct{}),
	}
}
132709
func NewConsumer(topicName string, subscriptionName string) Consumer {
	pubsubClient, err := newPubSubClient()

	if err != nil {
		log.Fatalf("Could not create PubSub client: %v", err)
	}

	topic := ensureTopic(pubsubClient, topicName)
	sub := ensureSubscription(pubsubClient, topic, subscriptionName)

	return &googlePubSubConsumer{Subscription: sub}
}