// Creates the Kafka topic for the channel if it does not already exist
973131
func setupTopicForChannel(retryOptions localconfig.Retry, haltChan chan struct{}, brokers []string, brokerConfig *sarama.Config, topicDetail *sarama.TopicDetail, channel channel) error {

	// requires Kafka v0.10.1.0 or higher
	if !brokerConfig.Version.IsAtLeast(sarama.V0_10_1_0) {
		return nil
	}

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

	retryMsg := fmt.Sprintf("Creating Kafka topic [%s] for channel [%s]",
		channel.topic(), channel.String())

	setupTopic := newRetryProcess(
		retryOptions,
		haltChan,
		channel,
		retryMsg,
		func() error {

			var err error
			clusterMembers := map[int32]*sarama.Broker{}
			var controllerId int32

			// loop through brokers to access metadata
			for _, address := range brokers {
				broker := sarama.NewBroker(address)
				err = broker.Open(brokerConfig)

				if err != nil {
					continue
				}

				var ok bool
				ok, err = broker.Connected()
				if !ok {
					continue
				}
				defer broker.Close()

				// metadata request which includes the topic
				var apiVersion int16
				if brokerConfig.Version.IsAtLeast(sarama.V0_11_0_0) {
					// use API version 4 to disable auto topic creation for
					// metadata requests
					apiVersion = 4
				} else {
					apiVersion = 1
				}
				metadata, err := broker.GetMetadata(&sarama.MetadataRequest{
					Version:                apiVersion,
					Topics:                 []string{channel.topic()},
					AllowAutoTopicCreation: false})

				if err != nil {
					continue
				}

				controllerId = metadata.ControllerID
				for _, broker := range metadata.Brokers {
					clusterMembers[broker.ID()] = broker
				}

				for _, topic := range metadata.Topics {
					if topic.Name == channel.topic() {
						if topic.Err != sarama.ErrUnknownTopicOrPartition {
							// auto create topics must be enabled so return
							return nil
						}
					}
				}
				break
			}

			// check to see if we got any metadata from any of the brokers in the list
			if len(clusterMembers) == 0 {
				return fmt.Errorf(
					"error creating topic [%s]; failed to retrieve metadata for the cluster",
					channel.topic())
			}

			// get the controller
			controller := clusterMembers[controllerId]
			err = controller.Open(brokerConfig)

			if err != nil {
				return err
			}

			var ok bool
			ok, err = controller.Connected()
			if !ok {
				return err
			}
			defer controller.Close()

			// create the topic
			req := &sarama.CreateTopicsRequest{
				Version: 0,
				TopicDetails: map[string]*sarama.TopicDetail{
					channel.topic(): topicDetail},
				Timeout: 3 * time.Second}
			resp := &sarama.CreateTopicsResponse{}
			resp, err = controller.CreateTopics(req)
			if err != nil {
				return err
			}

			// check the response
			if topicErr, ok := resp.TopicErrors[channel.topic()]; ok {
				// treat no error and topic exists error as success
				if topicErr.Err == sarama.ErrNoError ||
					topicErr.Err == sarama.ErrTopicAlreadyExists {
					return nil
				}
				if topicErr.Err == sarama.ErrInvalidTopic {
					// topic is invalid so abort
					logger.Warningf("[channel: %s] Failed to set up topic = %s",
						channel.topic(), topicErr.Err.Error())
					go func() {
						haltChan <- struct{}{}
					}()
				}
				return fmt.Errorf("error creating topic: [%s]",
					topicErr.Err.Error())
			}

			return nil
		})

	return setupTopic.retry()
}
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"
}
822586
func NewKafkaLogEmitter(config *KafkaLogEmitterConfig) (*KafkaLogEmitter, error) {
	encoder := kafkaavro.NewKafkaAvroEncoder(config.SchemaRegistryUrl)
	connector, err := siesta.NewDefaultConnector(config.ConnectorConfig)
	if err != nil {
		return nil, err
	}
	emitter := &KafkaLogEmitter{
		config:   config,
		producer: producer.NewKafkaProducer(config.ProducerConfig, producer.ByteSerializer, encoder.Encode, connector),
	}

	return emitter, nil
}
1162655
func newChannel(c *Connection, id uint16) *Channel {
	return &Channel{
		connection: c,
		id:         id,
		rpc:        make(chan message),
		consumers:  makeConsumers(),
		confirms:   newConfirms(),
		recv:       (*Channel).recvMethod,
		errors:     make(chan *Error, 1),
	}
}
424015
func (buf *SharedBuffer) NewChannel() SimpleChannel {
	ch := &sharedBufferChannel{
		in:  make(chan interface{}),
		out: make(chan interface{}),
		buf: queue.New(),
	}
	buf.in <- ch
	return ch
}
1259158
func newChannel(name string) *channel {
	return &channel{
		name:   name,
		modes:  new(ChanMode),
		nicks:  make(map[*nick]*ChanPrivs),
		lookup: make(map[string]*nick),
	}
}
75331
func NewTopic(topicName string, ctx *context, deleteCallback func(*Topic)) *Topic {
	t := &Topic{
		name:              topicName,
		channelMap:        make(map[string]*Channel),
		memoryMsgChan:     make(chan *Message, ctx.nsqd.getOpts().MemQueueSize),
		startChan:         make(chan int, 1),
		exitChan:          make(chan int),
		channelUpdateChan: make(chan int),
		ctx:               ctx,
		paused:            0,
		pauseChan:         make(chan int),
		deleteCallback:    deleteCallback,
		idFactory:         NewGUIDFactory(ctx.nsqd.getOpts().ID),
	}

	if strings.HasSuffix(topicName, "#ephemeral") {
		t.ephemeral = true
		t.backend = newDummyBackendQueue()
	} else {
		dqLogf := func(level diskqueue.LogLevel, f string, args ...interface{}) {
			opts := ctx.nsqd.getOpts()
			lg.Logf(opts.Logger, opts.LogLevel, lg.LogLevel(level), f, args...)
		}
		t.backend = diskqueue.New(
			topicName,
			ctx.nsqd.getOpts().DataPath,
			ctx.nsqd.getOpts().MaxBytesPerFile,
			int32(minValidMsgLength),
			int32(ctx.nsqd.getOpts().MaxMsgSize)+minValidMsgLength,
			ctx.nsqd.getOpts().SyncEvery,
			ctx.nsqd.getOpts().SyncTimeout,
			dqLogf,
		)
	}

	t.waitGroup.Wrap(t.messagePump)

	t.ctx.nsqd.Notify(t)

	return t
}
726834
func (a *adapter) Topic(res *hal.Response, strings ...string) error {
	for _, str := range strings {
		a.conn.SendRawf("TOPIC %s %s", res.Room(), str)
	}
	return nil
}
360506
func (c Channels) create(id uint32) *Channel {
	channel := &Channel{
		ID:       id,
		Links:    Channels{},
		Children: Channels{},
		Users:    Users{},
	}
	c[id] = channel
	return channel
}
71821
func (o *KeyServerMonitor) CreateKeyServerMonitorSEK(child *KeyServerMonitorSEK) *bambou.Error {

	return bambou.CurrentSession().CreateChild(o, child)
}
889421
func (api *Client) SetChannelTopic(channelID, topic string) (string, error) {
	return api.SetChannelTopicContext(context.Background(), channelID, topic)
}
40019
func (t *Topic) Produce(msg Message) error {
	userdata := newBuffer(msg.GetUserdata())
	native := newNativeMessage(msg)
	defer native.free()

	if C.Produce(t.handle, native.key, native.keyLen, native.payload, native.payloadLen, userdata) != 0 {
		defer freeBuffer(userdata)
		rspErr := ResponseError{
			Userdata: msg.GetUserdata(),
			Code:     int(C.GetLastError()),
		}
		if rspErr.Code == C.RD_KAFKA_RESP_ERR__QUEUE_FULL {
			t.client.Poll(-1)
		}
		return rspErr // ### return, error ###
	}
	return nil
}
1320317
func (t *Topic) GetChannel(channelName string) *Channel {
	t.Lock()
	channel, isNew := t.getOrCreateChannel(channelName)
	t.Unlock()

	if isNew {
		// update messagePump state
		select {
		case t.channelUpdateChan <- 1:
		case <-t.exitChan:
		}
	}

	return channel
}
684297
func createChannels(opts *Options) []wsstream.ChannelType {
	// open the requested channels, and always open the error channel
	channels := make([]wsstream.ChannelType, 5)
	channels[stdinChannel] = readChannel(opts.Stdin)
	channels[stdoutChannel] = writeChannel(opts.Stdout)
	channels[stderrChannel] = writeChannel(opts.Stderr)
	channels[errorChannel] = wsstream.WriteChannel
	channels[resizeChannel] = wsstream.ReadChannel
	return channels
}
895196
func (kinesis *Kinesis) CreateStream(StreamName string, ShardCount int) error {
	params := makeParams("CreateStream")
	requestParams := struct {
		StreamName string
		ShardCount int
	}{
		StreamName,
		ShardCount,
	}
	err := kinesis.query(params, requestParams, nil)
	if err != nil {
		return err
	}
	return nil
}
274564
func (c *Consumer) SubscribeTopics(topics []string, rebalanceCb RebalanceCb) (err error) {
	ctopics := C.rd_kafka_topic_partition_list_new(C.int(len(topics)))
	defer C.rd_kafka_topic_partition_list_destroy(ctopics)

	for _, topic := range topics {
		ctopic := C.CString(topic)
		defer C.free(unsafe.Pointer(ctopic))
		C.rd_kafka_topic_partition_list_add(ctopics, ctopic, C.RD_KAFKA_PARTITION_UA)
	}

	e := C.rd_kafka_subscribe(c.handle.rk, ctopics)
	if e != C.RD_KAFKA_RESP_ERR_NO_ERROR {
		return newError(e)
	}

	c.rebalanceCb = rebalanceCb
	c.handle.currAppRebalanceEnable = c.rebalanceCb != nil || c.appRebalanceEnable

	return nil
}
426063
func (c *Client) Publish(topic string, args []interface{}, kwargs map[string]interface{}) error {
	return c.Send(&Publish{
		Request:     NewID(),
		Options:     make(map[string]interface{}),
		Topic:       URI(topic),
		Arguments:   args,
		ArgumentsKw: kwargs,
	})
}