// StartConsumingChannel spawns receiving channel on AMQP queue
549529
func (b *AMQPCeleryBroker) StartConsumingChannel() error {
	channel, err := b.Consume(b.queue.Name, "", false, false, false, false, nil)
	if err != nil {
		return err
	}
	b.consumingChannel = channel
	return nil
}
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)
}
1345480
func (r *rpcServer) fetchActiveChannel(chanPoint wire.OutPoint) (
	*lnwallet.LightningChannel, error) {

	dbChan, err := r.server.chanDB.FetchChannel(chanPoint)
	if err != nil {
		return nil, err
	}

	// If the channel is successfully fetched from the database,
	// we create a fully populated channel state machine which
	// uses the db channel as backing storage.
	return lnwallet.NewLightningChannel(
		r.server.cc.wallet.Cfg.Signer, nil, dbChan, nil,
	)
}
1012735
func (eagerBroker *Broker) StartConsuming(consumerTag string, concurrency int, p iface.TaskProcessor) (bool, error) {
	return true, nil
}
745579
func consumeQueue(c *amqp.Channel, queue string, wg *sync.WaitGroup, done chan struct{}, out chan client.MessageSet) error {
	defer func() {
		log.With("queue", queue).Infoln("consuming complete")
		wg.Done()
	}()
	deliveries, err := c.Consume(queue, "transporter", false, false, false, false, nil)
	if err != nil {
		return err
	}
	for {
		select {
		case <-done:
			return nil
		case msg := <-deliveries:
			var result map[string]interface{}
			if jerr := json.NewDecoder(bytes.NewReader(msg.Body)).Decode(&result); jerr != nil {
				log.Errorf("unable to decode message to JSON, %s", jerr)
				continue
			}
			out <- client.MessageSet{
				Msg:  message.From(ops.Insert, queue, result),
				Mode: commitlog.Sync,
			}
			msg.Ack(false)
		}
	}
}
527472
func (conn *ProtoConnectionFields) StopConsumingPartition(topic string, partition int32, offset int64) error {
	return conn.multiplexer.stopConsumingPartition(topic, partition, offset, conn.name)
}
579856
func NewChannelQueue(capacity int) (Queue, error) {
	cap := int32(capacity)
	if capacity <= 0 {
		return nil, fmt.Errorf("Invalid size for queue: %d", capacity)
	}
	qChan := make(chan interface{}, cap)

	return &chanQueue{capacity: cap, qChan: qChan}, nil
}
333644
func (queue *redisQueue) StartConsuming(prefetchLimit int, pollDuration time.Duration) bool {
	if queue.deliveryChan != nil {
		return false // already consuming
	}

	// add queue to list of queues consumed on this connection
	if ok := queue.redisClient.SAdd(queue.queuesKey, queue.name); !ok {
		log.Panicf("rmq queue failed to start consuming %s", queue)
	}

	queue.prefetchLimit = prefetchLimit
	queue.pollDuration = pollDuration
	queue.deliveryChan = make(chan Delivery, prefetchLimit)
	atomic.StoreInt32(&queue.consumingStopped, 0)
	// log.Printf("rmq queue started consuming %s %d %s", queue, prefetchLimit, pollDuration)
	go queue.consume()
	return true
}
697509
func (conn *BytesManualConnectionStr) StartPostInitConsumer(topic string, partition int32, offset int64) (*sarama.PartitionConsumer, error) {
	multiplexer := conn.multiplexer
	multiplexer.WithFields(logging.Fields{"topic": topic}).Debugf("Post-init consuming started")

	if multiplexer.Consumer == nil || multiplexer.Consumer.SConsumer == nil {
		multiplexer.Warn("Unable to start post-init Consumer, client not available in the mux")
		return nil, nil
	}

	// Consumer that reads topic/partition/offset. Throws error if offset is 'in the future' (message with offset does not exist yet)
	partitionConsumer, err := multiplexer.Consumer.SConsumer.ConsumePartition(topic, partition, offset)
	if err != nil {
		return nil, err
	}
	multiplexer.Consumer.StartConsumerManualHandlers(partitionConsumer)

	return &partitionConsumer, nil
}
324395
func (api *Client) MarkIMChannel(channel, ts string) (err error) {
	return api.MarkIMChannelContext(context.Background(), channel, ts)
}
1007907
func (b *Broker) StopConsuming() {
	b.Broker.StopConsuming()

	b.stopReceiving()

	// Waiting for any tasks being processed to finish
	b.processingWG.Wait()

	// Waiting for the receiving goroutine to have stopped
	b.receivingWG.Wait()
}
11418
func (b *Broker) StartConsuming(consumerTag string, concurrency int, taskProcessor iface.TaskProcessor) {
	if b.retryFunc == nil {
		b.retryFunc = retry.Closure()
	}

	b.stopChan = make(chan int)
	b.retryStopChan = make(chan int)
}
859041
func (p *DefaultChannelClient) Open() error {
	channel, err := p.client.openChannel(p)
	if err != nil {
		return fmt.Errorf("Could not open AMQP channel (%s)", err)
	}

	p.channel = channel
	return nil
}
851942
func (api *Client) SetChannelPurpose(channelID, purpose string) (string, error) {
	return api.SetChannelPurposeContext(context.Background(), channelID, purpose)
}
371540
func (c *Cast) StartTabMirroring(sinkName string) (*gcdmessage.ChromeResponse, error) {
	var v CastStartTabMirroringParams
	v.SinkName = sinkName
	return c.StartTabMirroringWithParams(&v)
}
97251
func (f *asyncFlusher) start(numParallel int) {
	f.chunks = make(chan chunk)
	for i := 0; i < numParallel; i++ {
		f.wg.Add(1)
		go func() {
			defer f.wg.Done()
			f.uploaderLoop()
		}()
	}
}
625030
func (md *MDServerMemory) StartImplicitTeamMigration(
	ctx context.Context, id tlf.ID) (err error) {
	md.lock.Lock()
	defer md.lock.Unlock()
	md.iTeamMigrationLocks[id] = true
	return nil
}