// channel_producer serves the ProduceChannel channel
771249
func channelProducer(p *Producer) {

	for m := range p.produceChannel {
		err := p.produce(m, C.RD_KAFKA_MSG_F_BLOCK, nil)
		if err != nil {
			m.TopicPartition.Error = err
			p.events <- m
		}
	}

	p.handle.terminatedChan <- "channelProducer"
}
1053446
func (d *API) ProducersFor(mediaTypes []string) map[string]runtime.Producer {
	result := make(map[string]runtime.Producer)
	for _, mt := range mediaTypes {
		if producer, ok := d.producers[mt]; ok {
			result[mt] = producer
		}
	}
	return result
}
606990
func (f *RBForwarder) Produce(data []byte, opts map[string]interface{}, opaque interface{}) error {
	if atomic.LoadUint32(&f.working) == 0 {
		return errors.New("Forwarder has been closed")
	}

	seq := f.currentProducedID
	f.currentProducedID++
	m := utils.NewMessage()
	r := Report{
		seq:    seq,
		Opaque: opaque,
	}
	m.PushPayload(data)
	m.Opts.MSet(opts)
	m.Reports.Push(r)

	f.p.input <- m

	return nil
}
57676
func (prod *S3) Produce(workers *sync.WaitGroup) {
	prod.AddMainWorker(workers)
	sess, err := session.NewSessionWithOptions(session.Options{
		Config:            *prod.config,
		SharedConfigState: session.SharedConfigEnable,
	})
	if err != nil {
		prod.Logger.WithError(err).Error("Failed to create session")
	}

	if prod.assumeRole != "" {
		creds := stscreds.NewCredentials(sess, prod.assumeRole)
		prod.client = s3.New(sess, &aws.Config{Credentials: creds})
	} else {
		prod.client = s3.New(sess)
	}

	prod.TickerMessageControlLoop(prod.bufferMessage, prod.flushFrequency, prod.sendBatchOnTimeOut)
}
225952
func (prod *InfluxDB) Produce(workers *sync.WaitGroup) {
	prod.BatchMessageLoop(workers, prod.sendBatch)
}
1131595
func (prod *DirectProducer) enqueuePanicHandling(msg *Message) {
	if r := recover(); r != nil {
		prod.Logger.Error("Recovered a panic during producer enqueue: ", r)
		prod.Logger.Error("Producer: ", prod.id, "State: ", prod.GetState(),
			", Router: ", StreamRegistry.GetStreamName(msg.GetStreamID()))
		prod.TryFallback(msg)
	}
}
696138
func (prod *AwsCloudwatchLogs) Produce(workers *sync.WaitGroup) {
	prod.initService()
	if err := prod.create(); err != nil {
		prod.Logger.Errorf("could not create group:%q stream:%q error was: %q", prod.group, prod.stream, err)
	} else {
		prod.setToken()
	}
	prod.BatchMessageLoop(workers, prod.sendBatch)
}
138429
func MakeChannelCreationTransaction(channelID string, signer crypto.LocalSigner, conf *genesisconfig.Profile) (*cb.Envelope, error) {
	template, err := DefaultConfigTemplate(conf)
	if err != nil {
		return nil, errors.WithMessage(err, "could not generate default config template")
	}
	return MakeChannelCreationTransactionFromTemplate(channelID, signer, conf, template)
}
638319
func ChannelCreationBlockToGenesisBlock(block *common.Block) (*common.Block, error) {
	if block == nil {
		return nil, errors.New("nil block")
	}
	env, err := utils.ExtractEnvelope(block, 0)
	if err != nil {
		return nil, err
	}
	payload, err := utils.ExtractPayload(env)
	if err != nil {
		return nil, err
	}
	block.Data.Data = [][]byte{payload.Data}
	block.Header.DataHash = block.Data.Hash()
	block.Header.Number = 0
	block.Header.PreviousHash = nil
	metadata := &common.BlockMetadata{
		Metadata: make([][]byte, 4),
	}
	block.Metadata = metadata
	metadata.Metadata[common.BlockMetadataIndex_LAST_CONFIG] = utils.MarshalOrPanic(&common.Metadata{
		Value: utils.MarshalOrPanic(&common.LastConfig{Index: 0}),
		// This is a genesis block, peer never verify this signature because we can't bootstrap
		// trust from an earlier block, hence there are no signatures here.
	})
	return block, nil
}
669858
func (t *Topic) getOrCreateChannel(channelName string) (*Channel, bool) {
	channel, ok := t.channelMap[channelName]
	if !ok {
		deleteCallback := func(c *Channel) {
			t.DeleteExistingChannel(c.name)
		}
		channel = NewChannel(t.name, channelName, t.ctx, deleteCallback)
		t.channelMap[channelName] = channel
		t.ctx.nsqd.logf(LOG_INFO, "TOPIC(%s): new channel(%s)", t.name, channel.name)
		return channel, true
	}
	return channel, false
}
1356662
func (p *asyncProducer) newBrokerProducer(broker *Broker) *brokerProducer {
	var (
		input     = make(chan *ProducerMessage)
		bridge    = make(chan *produceSet)
		responses = make(chan *brokerProducerResponse)
	)

	bp := &brokerProducer{
		parent:         p,
		broker:         broker,
		input:          input,
		output:         bridge,
		responses:      responses,
		buffer:         newProduceSet(p),
		currentRetries: make(map[string]map[int32]error),
	}
	go withRecover(bp.run)

	// minimal bridge to make the network response `select`able
	go withRecover(func() {
		for set := range bridge {
			request := set.buildRequest()

			response, err := broker.Produce(request)

			responses <- &brokerProducerResponse{
				set: set,
				err: err,
				res: response,
			}
		}
		close(responses)
	})

	if p.conf.Producer.Retry.Max <= 0 {
		bp.abandoned = make(chan struct{})
	}

	return bp
}
657858
func (q Query) ToChannel(result chan<- interface{}) {
	next := q.Iterate()

	for item, ok := next(); ok; item, ok = next() {
		result <- item
	}

	close(result)
}
1004023
func (p *Producer) Close() {
	// Wait for poller() (signaled by closing pollerTermChan)
	// and channel_producer() (signaled by closing ProduceChannel)
	close(p.pollerTermChan)
	close(p.produceChannel)
	p.handle.waitTerminated(2)

	close(p.events)

	p.handle.cleanup()

	C.rd_kafka_destroy(p.handle.rk)
}
428137
func setupProducerForChannel(retryOptions localconfig.Retry, haltChan chan struct{}, brokers []string, brokerConfig *sarama.Config, channel channel) (sarama.SyncProducer, error) {
	var err error
	var producer sarama.SyncProducer

	logger.Infof("[channel: %s] Setting up the producer for this channel...", channel.topic())

	retryMsg := "Connecting to the Kafka cluster"
	setupProducer := newRetryProcess(retryOptions, haltChan, channel, retryMsg, func() error {
		producer, err = sarama.NewSyncProducer(brokers, brokerConfig)
		return err
	})

	return producer, setupProducer.retry()
}
1255234
func (prod *ElasticSearch) Produce(workers *sync.WaitGroup) {
	defer prod.WorkerDone()

	// create all indexes that are not time based
	for _, item := range prod.indexMap {
		if !item.useTimeIndex {
			prod.createIndexIfRequired(item.name, item.settings)
		}
	}

	prod.AddMainWorker(workers)
	prod.BatchMessageLoop(workers, func() core.AssemblyFunc { return prod.submitMessages })
}
118041
func (b *Broker) Produce(request *ProduceRequest) (*ProduceResponse, error) {
	var (
		response *ProduceResponse
		err      error
	)

	if request.RequiredAcks == NoResponse {
		err = b.sendAndReceive(request, nil)
	} else {
		response = new(ProduceResponse)
		err = b.sendAndReceive(request, response)
	}

	if err != nil {
		return nil, err
	}

	return response, nil
}
607955
func (m *ChannelDeMultiplexer) AddChannel(predicate common.MessageAcceptor) chan interface{} {
	m.lock.Lock()
	defer m.lock.Unlock()
	ch := &channel{ch: make(chan interface{}, 10), pred: predicate}
	m.channels = append(m.channels, ch)
	return ch.ch
}