// NewConsumer constructs a new consumer configuration.
// The event channel will be passed a collection of rows generated from all of the events in a single block
// It will be closed by the consumer when it is finished
773500
func NewConsumer(cfg *config.VentConfig, log *logger.Logger, eventChannel chan types.EventData) *Consumer {
	return &Consumer{
		Config:        cfg,
		Log:           log,
		Closing:       false,
		EventsChannel: eventChannel,
	}
}
162833
func New(streamName string, opts ...Option) (*Consumer, error) {
	if streamName == "" {
		return nil, fmt.Errorf("must provide stream name")
	}

	// new consumer with no-op checkpoint, counter, and logger
	c := &Consumer{
		streamName:               streamName,
		initialShardIteratorType: kinesis.ShardIteratorTypeLatest,
		checkpoint:               &noopCheckpoint{},
		counter:                  &noopCounter{},
		logger: &noopLogger{
			logger: log.New(ioutil.Discard, "", log.LstdFlags),
		},
	}

	// override defaults
	for _, opt := range opts {
		opt(c)
	}

	// default client if none provided
	if c.client == nil {
		newSession, err := session.NewSession(aws.NewConfig())
		if err != nil {
			return nil, err
		}
		c.client = kinesis.New(newSession)
	}

	return c, nil
}
470361
func (queue *Queue) AddConsumer(name string) (c *Consumer, err error) {
	c = &Consumer{Name: name, Queue: queue}
	//check uniqueness and start heartbeat
	added, err := queue.redisClient.SAdd(queueWorkersKey(queue.Name), name).Result()
	if err != nil {
		return nil, err
	}
	if added == 0 {
		if queue.isActiveConsumer(name) {
			return nil, fmt.Errorf("consumer with this name is already active")
		}
	}
	c.startHeartbeat()
	return c, nil
}
443150
func New() *Consumer {
	return &Consumer{
		Logger:   nsqlog.Logger,
		LogLevel: nsqlog.LogLevel,
		handlers: make(map[topicChan]*queue),
	}
}
505452
func NewConverter() *Converter {
	c := &Converter{
		conversionFuncs:          map[typePair]reflect.Value{},
		generatedConversionFuncs: map[typePair]reflect.Value{},
		defaultingFuncs:          map[reflect.Type]reflect.Value{},
		defaultingInterfaces:     map[reflect.Type]interface{}{},
		nameFunc:                 func(t reflect.Type) string { return t.Name() },
		structFieldDests:         map[typeNamePair][]typeNamePair{},
		structFieldSources:       map[typeNamePair][]typeNamePair{},

		inputFieldMappingFuncs: map[reflect.Type]FieldMappingFunc{},
		inputDefaultFlags:      map[reflect.Type]FieldMatchingFlags{},
	}
	c.RegisterConversionFunc(ByteSliceCopy)
	return c
}
1164053
func (eventBroadcaster *eventBroadcasterImpl) NewRecorder(scheme *runtime.Scheme, source v1.EventSource) EventRecorder {
	return &recorderImpl{scheme, source, eventBroadcaster.Broadcaster, clock.RealClock{}}
}
662984
func (ref *Consumer) StartConsumerHandlers() {
	config := ref.Config
	config.Logger.Info("Starting message handlers for new consumer ...")
	// if required, start reading from the notifications channel
	if config.ConsumerConfig().Group.Return.Notifications {
		go ref.notificationHandler(ref.Consumer.Notifications())
	}

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

	// start the message handler
	go ref.messageHandler(ref.Consumer.Messages())
}
1175957
func (o *CountdownAPI) RegisterConsumer(mediaType string, consumer runtime.Consumer) {
	o.customConsumers[mediaType] = consumer
}
783850
func InitConsumer(topic string) chan []byte {
	consumer, err := bus.StartConsumer(topic)
	if err != nil {
		fmt.Println(err)
	}
	return consumer
}
427755
func NewConsumerConf(topic string, partition int32) ConsumerConf {
	return ConsumerConf{
		Topic:          topic,
		Partition:      partition,
		RequestTimeout: time.Millisecond * 50,
		RetryLimit:     -1,
		RetryWait:      time.Millisecond * 50,
		RetryErrLimit:  10,
		RetryErrWait:   time.Millisecond * 500,
		MinFetchSize:   1,
		MaxFetchSize:   2000000,
		StartOffset:    StartOffsetOldest,
		Logger:         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
}
381552
func (db *DB) newBuilder(executor Executor) Builder {
	builderFunc, ok := BuilderFuncMap[db.driverName]
	if !ok {
		builderFunc = NewStandardBuilder
	}
	return builderFunc(db, executor)
}
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,
	}
}
1363089
func newWatcher(c chan<- EventInfo) watcher {
	t := &trg{
		s:      make(chan struct{}, 1),
		pthLkp: make(map[string]*watched, 0),
		c:      c,
	}
	t.t = newTrigger(t.pthLkp)
	if err := t.t.Init(); err != nil {
		t.Close()
		return watcherStub{fmt.Errorf("failed setting up watcher: %v", err)}
	}
	go t.monitor()
	return t
}
933222
func (server *Server) NewWorker(consumerTag string, concurrency int) *Worker {
	return &Worker{
		server:      server,
		ConsumerTag: consumerTag,
		Concurrency: concurrency,
		Queue:       "",
	}
}
681804
func NewConsumerFromClient(client Client) (Consumer, error) {
	// For clients passed in by the client, ensure we don't
	// call Close() on it.
	cli := &nopCloserClient{client}
	return newConsumer(cli)
}
535322
func NewDeliverService(conf *Config) (DeliverService, error) {
	ds := &deliverServiceImpl{
		conf:           conf,
		blockProviders: make(map[string]blocksprovider.BlocksProvider),
	}
	if err := ds.validateConfiguration(); err != nil {
		return nil, err
	}
	return ds, nil
}