// SubConsumers returns the other UI consumers for this engine.
423423
func (e *SaltpackEncrypt) SubConsumers() []libkb.UIConsumer {
	// Note that potentially KeyfinderHook might return a different UIConsumer depending on its arguments,
	// which might make this call problematic, but all the hooks currently in use are not doing that.
	return []libkb.UIConsumer{
		e.newKeyfinderHook(libkb.SaltpackRecipientKeyfinderArg{}),
	}
}
511533
func (s *NotificationWithSubscribers) SetSubscribers(v []*Subscriber) *NotificationWithSubscribers {
	s.Subscribers = v
	return s
}
372172
func (o Middleware) SubscribeInterceptor(opts pubsub.HandlerOptions, next pubsub.MsgHandler) pubsub.MsgHandler {
	return func(ctx context.Context, m pubsub.Msg) error {
		start := time.Now()
		err := next(ctx, m)

		pubsubHandled.WithLabelValues(
			opts.Topic,
			opts.ServiceName,
			strconv.FormatBool(err == nil),
		).Inc()

		subscriberSize.WithLabelValues(
			opts.Topic,
			opts.ServiceName,
		).Add(float64(len(m.Data)))

		subscribeDurationsHistogram.WithLabelValues(
			opts.Topic,
			opts.ServiceName,
		).Observe(time.Now().Sub(start).Seconds())
		return err
	}
}
998344
func (it *And) SubIterators() []graph.Iterator {
	iters := make([]graph.Iterator, 0, len(it.internalIterators)+1)
	if it.primaryIt != nil {
		iters = append(iters, it.primaryIt)
	}
	iters = append(iters, it.internalIterators...)
	return iters
}
1271386
func (s *DescribeSubscribersForNotificationOutput) SetSubscribers(v []*Subscriber) *DescribeSubscribersForNotificationOutput {
	s.Subscribers = v
	return s
}
1029735
func JSONConsumer() Consumer {
	return ConsumerFunc(func(reader io.Reader, data interface{}) error {
		dec := json.NewDecoder(reader)
		dec.UseNumber() // preserve number formats
		return dec.Decode(data)
	})
}
1188399
func Subscribers() string {
	subs := make(map[string]bool) // no duplicates

	// get tags all clients subscribed to
	for i := range subscribers {
		s := subscribers[i].subscriptions.ToSlice()
		for j := range s {
			for k := range s[j] {
				subs[s[j][k]] = true
			}
		}
	}

	// slice it
	subSlice := []string{}
	for k, _ := range subs {
		subSlice = append(subSlice, k)
	}

	return strings.Join(subSlice, " ")
}
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
}
258757
func (o *TodoListAPI) ConsumersFor(mediaTypes []string) map[string]runtime.Consumer {

	result := make(map[string]runtime.Consumer)
	for _, mt := range mediaTypes {
		switch mt {

		case "application/io.goswagger.examples.todo-list.v1+json":
			result["application/io.goswagger.examples.todo-list.v1+json"] = o.JSONConsumer

		}

		if c, ok := o.customConsumers[mt]; ok {
			result[mt] = c
		}
	}
	return result

}
1136827
func (pm *Manager) SubscribeEvents(buffer int, watchEvents ...Event) (eventCh <-chan interface{}, cancel func()) {
	topic := func(i interface{}) bool {
		observed, ok := i.(Event)
		if !ok {
			panic(fmt.Sprintf("unexpected type passed to event channel: %v", reflect.TypeOf(i)))
		}
		for _, e := range watchEvents {
			if e.matches(observed) {
				return true
			}
		}
		// If no specific events are specified always assume a matched event
		// If some events were specified and none matched above, then the event
		// doesn't match
		return watchEvents == nil
	}
	ch := pm.publisher.SubscribeTopicWithBuffer(topic, buffer)
	cancelFunc := func() { pm.publisher.Evict(ch) }
	return ch, cancelFunc
}
440932
func (it *Not) SubIterators() []graph.Iterator {
	return []graph.Iterator{it.primaryIt, it.allIt}
}
717267
func (r *Router) Subrouter(mws ...Middleware) *Router {
	nr := &Router{
		parent:           r,
		hostrm:           r.hostrm,
		pattern:          r.pattern,
		middlewares:      Middlewares{},
		namedMiddlewares: make(map[string]Middlewares),
		host:             r.host,
		pool:             newCtxPool(),
		routes:           []*route{},
		subrouters:       []*Router{},
	}
	nr.Use(mws...)
	r.subrouters = append(r.subrouters, nr)
	return nr
}
160529
func (o Middleware) SubscribeInterceptor(opts pubsub.HandlerOptions, next pubsub.MsgHandler) pubsub.MsgHandler {
	return func(ctx context.Context, m pubsub.Msg) error {
		ctx = context.WithValue(ctx, "x-audit-user", m.Metadata["x-audit-user"])
		ctx = metadata.NewIncomingContext(ctx, metadata.New(m.Metadata))
		return next(ctx, m)
	}
}
929545
func NewConsumer(addr string, consumer gonzo.Consumer) *Consumer {
	return &Consumer{
		address:        addr,
		consumer:       consumer,
		customHandlers: make(map[string]func([]byte) (*Response, error)),
		close:          make(chan struct{}),
		closed:         make(chan struct{}),
	}
}
678692
func (r *rpcServer) SubscribeInvoices(req *lnrpc.InvoiceSubscription,
	updateStream lnrpc.Lightning_SubscribeInvoicesServer) error {

	invoiceClient := r.server.invoices.SubscribeNotifications(
		req.AddIndex, req.SettleIndex,
	)
	defer invoiceClient.Cancel()

	for {
		select {
		case newInvoice := <-invoiceClient.NewInvoices:
			rpcInvoice, err := invoicesrpc.CreateRPCInvoice(
				newInvoice, activeNetParams.Params,
			)
			if err != nil {
				return err
			}

			if err := updateStream.Send(rpcInvoice); err != nil {
				return err
			}

		case settledInvoice := <-invoiceClient.SettledInvoices:
			rpcInvoice, err := invoicesrpc.CreateRPCInvoice(
				settledInvoice, activeNetParams.Params,
			)
			if err != nil {
				return err
			}

			if err := updateStream.Send(rpcInvoice); err != nil {
				return err
			}

		case <-r.quit:
			return nil
		}
	}
}
1002765
func XMLConsumer() Consumer {
	return ConsumerFunc(func(reader io.Reader, data interface{}) error {
		dec := xml.NewDecoder(reader)
		return dec.Decode(data)
	})
}
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
}