// NewQueue creates a new publish/subscribe queue which supports watchers.
// The channels that it will create for subscriptions will have the buffer
// size specified by buffer.
185625
func NewQueue(options ...func(*Queue) error) *Queue {
	// Create a queue with the default values
	q := &Queue{
		sinkGen:      &dropErrClosedChanGen{},
		broadcast:    events.NewBroadcaster(),
		cancelFuncs:  make(map[events.Sink]func()),
		limit:        0,
		closeOutChan: false,
	}

	for _, option := range options {
		err := option(q)
		if err != nil {
			panic(fmt.Sprintf("Failed to apply options to queue: %s", err))
		}
	}

	return q
}
546004
func NewEventQueueBuffered(name string, numBufferedEvents int) *EventQueue {
	log.WithFields(logrus.Fields{
		"name":              name,
		"numBufferedEvents": numBufferedEvents,
	}).Debug("creating new EventQueue")
	return &EventQueue{
		name: name,
		// Up to numBufferedEvents can be Enqueued until Enqueueing blocks.
		events: make(chan *Event, numBufferedEvents),
		close:  make(chan struct{}),
		drain:  make(chan struct{}),
	}

}
1047133
func (m *MockAgentConnector) QueueSubscribe(ctx context.Context, name, subject, group string, output chan *ConnectorMessage) error {
	ret := m.ctrl.Call(m, "QueueSubscribe", ctx, name, subject, group, output)
	ret0, _ := ret[0].(error)
	return ret0
}
95702
func NewQueue() Queue {
	q := Queue{
		jobs: make([]*Package, 0),
		lock: new(sync.RWMutex),
	}

	return q
}
1066929
func (s *Wrapper) NewWatcher(ctx context.Context, watch Watch) (Watcher, error) {
	if err := s.GetReadError(); err != nil {
		return nil, trace.Wrap(err)
	}
	return s.backend.NewWatcher(ctx, watch)
}
1226063
func (c Client) New(params *stripe.SubscriptionParams) (*stripe.Subscription, error) {
	sub := &stripe.Subscription{}
	err := c.B.Call(http.MethodPost, "/v1/subscriptions", c.Key, params, sub)
	return sub, err
}
493835
func (s *GossipStateProviderImpl) queueNewMessage(msg *proto.GossipMessage) {
	if !bytes.Equal(msg.Channel, []byte(s.chainID)) {
		logger.Warning("Received enqueue for channel",
			string(msg.Channel), "while expecting channel", s.chainID, "ignoring enqueue")
		return
	}

	dataMsg := msg.GetDataMsg()
	if dataMsg != nil {
		if err := s.addPayload(dataMsg.GetPayload(), NonBlocking); err != nil {
			logger.Warningf("Block [%d] received from gossip wasn't added to payload buffer: %v", dataMsg.Payload.SeqNum, err)
			return
		}

	} else {
		logger.Debug("Gossip message received is not of data message type, usually this should not happen.")
	}
}
449240
func New(chanDB *channeldb.DB) *ChannelNotifier {
	return &ChannelNotifier{
		ntfnServer: subscribe.NewServer(),
		chanDB:     chanDB,
	}
}
435108
func (b *sqsBroker) Subscribe(queueName string, h broker.Handler, opts ...broker.SubscribeOption) (broker.Subscriber, error) {
	queueURL, err := b.urlFromQueueName(queueName)
	if err != nil {
		return nil, err
	}

	options := broker.SubscribeOptions{
		AutoAck: true,
		Queue:   queueName,
		Context: context.Background(),
	}

	for _, o := range opts {
		o(&options)
	}

	subscriber := &subscriber{
		options:   options,
		URL:       queueURL,
		queueName: queueName,
		svc:       b.svc,
		exit:      make(chan bool),
	}
	go subscriber.run(h)

	return subscriber, nil
}
1206142
func NewLimitQueue(dst events.Sink, limit uint64) *LimitQueue {
	eq := LimitQueue{
		dst:    dst,
		events: list.New(),
		limit:  limit,
		full:   make(chan struct{}),
	}

	eq.cond = sync.NewCond(&eq.mu)
	go eq.run()
	return &eq
}
1306484
func (connection *redisConnection) OpenQueue(name string) Queue {
	connection.redisClient.SAdd(queuesKey, name)
	queue := newQueue(name, connection.Name, connection.queuesKey, connection.redisClient)
	return queue
}
1361051
func newChannelCache(n int) *channelCache {
	return &channelCache{
		n:        n,
		channels: make(map[uint64]ChannelEdge),
	}
}
510853
func NewPQueue(pqType PQType) *PQueue {
	var cmp func(int, int) bool

	if pqType == MAXPQ {
		cmp = max
	} else {
		cmp = min
	}

	items := make([]*item, 1)
	items[0] = nil // Heap queue first element should always be nil

	return &PQueue{
		items:      items,
		elemsCount: 0,
		comparator: cmp,
	}
}
831573
func (w *watchQueue) Enqueue(key string, kvp *kvdb.KVPair, err error) {
	w.m.Lock()
	w.updates.PushBack(&watchUpdate{key: key, kvp: kvp, err: err})
	w.cv.Signal()
	w.m.Unlock()
}
348398
func (r *marathonClient) Queue() (*Queue, error) {
	var queue *Queue
	err := r.apiGet(marathonAPIQueue, nil, &queue)
	if err != nil {
		return nil, err
	}
	return queue, nil
}
415191
func newDownloadQueue(manifest *tq.Manifest, remote string, options ...tq.Option) *tq.TransferQueue {
	return tq.NewTransferQueue(tq.Download, manifest, remote, append(options,
		tq.RemoteRef(currentRemoteRef()),
	)...)
}
207479
func (server *Server) NewCustomQueueWorker(consumerTag string, concurrency int, queue string) *Worker {
	return &Worker{
		server:      server,
		ConsumerTag: consumerTag,
		Concurrency: concurrency,
		Queue:       queue,
	}
}