// NewConsumer creates a new Consumer using the given client and config.
// The message processing logic is passed via strategy.
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{}),
	}
}
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)
		}
	}
}
1209719
func (c *Client) NewBuilder() *Builder {
	c.ensureTransmission()
	c.ensureBuilder()
	return c.builder.Clone()
}
501284
func NewConfig(consumerKey, consumerSecret string) *Config {
	return &Config{
		ConsumerKey:    consumerKey,
		ConsumerSecret: consumerSecret,
	}
}
38607
func GetConsumer(clientID string, clientSecret string) *oauth.Consumer {
	return oauth.NewConsumer(
		clientID,
		clientSecret,
		oauth.ServiceProvider{
			RequestTokenUrl:   YahooRequestTokenURL,
			AuthorizeTokenUrl: YahooAuthTokenURL,
			AccessTokenUrl:    YahooGetTokenURL,
		})
}
82473
func NewAdminClientFromConsumer(c *Consumer) (a *AdminClient, err error) {
	if c.handle.rk == nil {
		return nil, newErrorFromString(ErrInvalidArg, "Can't derive AdminClient from closed consumer")
	}

	a = &AdminClient{}
	a.handle = &c.handle
	a.isDerived = true
	return a, nil
}
1128182
func NewConsumer(q *Queue, opts ...ConsumerOpt) *Consumer {
	c := &Consumer{
		q:          q,
		deliveries: make(chan amqp.Delivery),
		errs:       make(chan error, 100),
		stop:       make(chan struct{}),
	}
	for _, o := range opts {
		o(c)
	}
	return c
}
1049892
func (queue *redisQueue) AddBatchConsumer(tag string, batchSize int, consumer BatchConsumer) string {
	return queue.AddBatchConsumerWithTimeout(tag, batchSize, defaultBatchTimeout, consumer)
}
1175957
func (o *CountdownAPI) RegisterConsumer(mediaType string, consumer runtime.Consumer) {
	o.customConsumers[mediaType] = consumer
}
21649
func NewConsumerGroup(brokers []string, topic string, group string, offset int64) (*ConsumerGroup, error) {
	return newConsumer(brokers, topic, group, offset, false)
}
1173135
func (ai *Amqp) CreateConsumer(exchange, key, kind, queue string, durable bool) (<-chan amqp.Delivery, error) {
	if err := ai.WithExchange(exchange, kind, durable); err != nil {
		return nil, err
	}

	q, err := ai.ch.QueueDeclare(queue, durable, false, false, false, nil)
	if err != nil {
		return nil, err
	}

	if err := ai.ch.QueueBind(q.Name, key, exchange, false, nil); err != nil {
		return nil, err
	}

	return ai.ch.Consume(q.Name, "", false, false, false, false, nil)
}
1167085
func NewConsumer(config *Config, wg *sync.WaitGroup) (*Consumer, error) {
	if config.Debug {
		config.Logger.SetLevel(logging.DebugLevel)
	}
	config.Logger.Debug("entering NewConsumer ...")
	if err := config.ValidateConsumerConfig(); err != nil {
		return nil, err
	}
	config.Logger.Debugf("Consumer config: %#v", config)

	// set consumer config params
	config.ConsumerConfig().Group.Return.Notifications = config.RecvNotification
	config.ProducerConfig().Consumer.Return.Errors = config.RecvError
	config.ConsumerConfig().Consumer.Offsets.Initial = config.InitialOffset

	cClient, err := cluster.NewClient(config.Brokers, config.Config)
	if err != nil {
		return nil, err
	}

	config.Logger.Debug("new client created successfully ...")

	consumer, err := cluster.NewConsumerFromClient(cClient, config.GroupID, config.Topics)
	if err != nil {
		return nil, err
	}

	sConsumer, err := sarama.NewConsumerFromClient(cClient)
	if err != nil {
		return nil, err
	}

	csmr := &Consumer{
		Logger:       config.Logger,
		Config:       config,
		SConsumer:    sConsumer,
		Consumer:     consumer,
		closed:       false,
		closeChannel: make(chan struct{}),
	}

	// if there is a "waitgroup" arg then use it
	if wg != nil {
		csmr.xwg = wg
		csmr.xwg.Add(1)
	}

	return csmr, nil
}
1220899
func (f *Factory) NewBuilder() *resource.Builder {
	mapper, typer := f.Object()

	return resource.NewBuilder(mapper, typer, resource.ClientMapperFunc(f.ClientForMapping), f.Decoder(true))
}
1120180
func NewConsumer(config *ConsumerConfig) *Consumer {
	if err := config.Validate(); err != nil {
		panic(err)
	}

	Infof(config.Consumerid, "Creating new consumer with configuration: %s", config)
	c := &Consumer{
		config:                         config,
		unsubscribe:                    make(chan bool),
		closeFinished:                  make(chan bool),
		topicPartitionsAndBuffers:      make(map[TopicAndPartition]*messageBuffer),
		topicRegistry:                  make(map[string]map[int32]*partitionTopicInfo),
		connectChannels:                make(chan bool),
		disconnectChannelsForPartition: make(chan TopicAndPartition),
		workerManagers:                 make(map[TopicAndPartition]*WorkerManager),
		stopStreams:                    make(chan bool),
		close:                          make(chan bool),
	}

	if err := c.config.Coordinator.Connect(); err != nil {
		panic(err)
	}
	if err := c.config.LowLevelClient.Initialize(); err != nil {
		panic(err)
	}
	c.metrics = newConsumerMetrics(c.String(), config.MetricsPrefix)
	c.fetcher = newConsumerFetcherManager(c.config, c.disconnectChannelsForPartition, c.metrics)

	go func() {
		<-c.close
		if !c.isShuttingdown {
			c.Close()
		}
	}()

	return c
}
809256
func TextConsumer() Consumer {
	return ConsumerFunc(func(reader io.Reader, data interface{}) error {
		if reader == nil {
			return errors.New("TextConsumer requires a reader") // early exit
		}

		buf := new(bytes.Buffer)
		_, err := buf.ReadFrom(reader)
		if err != nil {
			return err
		}
		b := buf.Bytes()

		// If the buffer is empty, no need to unmarshal it, which causes a panic.
		if len(b) == 0 {
			data = ""
			return nil
		}

		if tu, ok := data.(encoding.TextUnmarshaler); ok {
			err := tu.UnmarshalText(b)
			if err != nil {
				return fmt.Errorf("text consumer: %v", err)
			}

			return nil
		}

		t := reflect.TypeOf(data)
		if data != nil && t.Kind() == reflect.Ptr {
			v := reflect.Indirect(reflect.ValueOf(data))
			if t.Elem().Kind() == reflect.String {
				v.SetString(string(b))
				return nil
			}
		}

		return fmt.Errorf("%v (%T) is not supported by the TextConsumer, %s",
			data, data, "can be resolved by supporting TextUnmarshaler interface")
	})
}
195052
func NewClient(config *Config, partitioner string) (sarama.Client, error) {
	config.Logger.Debug("Creating new consumer")
	if err := config.ValidateAsyncProducerConfig(); err != nil {
		return nil, err
	}

	config.SetSendSuccess(true)
	config.SetSuccessChan(make(chan *ProducerMessage))
	config.SetSendError(true)
	config.SetErrorChan(make(chan *ProducerError))
	// Required acks will be set in sync/async producer
	config.RequiredAcks = AcksUnset

	// set other Producer config params
	config.ProducerConfig().Producer.Return.Successes = config.SendSuccess
	config.ProducerConfig().Producer.Return.Errors = config.SendError

	// set partitioner
	switch partitioner {
	case Hash:
		config.ProducerConfig().Producer.Partitioner = sarama.NewHashPartitioner
	case Random:
		config.ProducerConfig().Producer.Partitioner = sarama.NewRandomPartitioner
	case Manual:
		config.ProducerConfig().Producer.Partitioner = sarama.NewManualPartitioner
	default:
		// Hash partitioner is set as default
		config.ProducerConfig().Producer.Partitioner = sarama.NewHashPartitioner
	}

	config.Logger.Debugf("AsyncProducer config: %#v", config)

	sClient, err := sarama.NewClient(config.Brokers, &config.Config.Config)
	if err != nil {
		fmt.Printf("Error creating consumer client %v", err)
		return nil, err
	}

	return sClient, nil
}
822201
func (ref *Consumer) StartConsumerManualHandlers(partitionConsumer sarama.PartitionConsumer) {
	config := ref.Config
	config.Logger.Info("Starting message handlers for new manual consumer ...")

	// if required, start reading from the errors channel
	if config.ProducerConfig().Consumer.Return.Errors {
		go ref.manualErrorHandler(partitionConsumer.Errors())
	}

	// start the message handler
	go ref.messageHandler(partitionConsumer.Messages())
}