// NewSUBSCRIBE creates and returns a SUBSCRIBE Packet.
436688
func NewSUBSCRIBE(opts *SUBSCRIBEOptions) (Packet, error) {
	// Initialize the options.
	if opts == nil {
		opts = &SUBSCRIBEOptions{}
	}

	// Validate the options.
	if err := opts.validate(); err != nil {
		return nil, err
	}

	// Create a SUBSCRIBE Packet.
	p := &SUBSCRIBE{
		PacketID: opts.PacketID,
		SubReqs:  opts.SubReqs,
	}

	// Set the variable header to the Packet.
	p.setVariableHeader()

	// Set the payload to the Packet.
	p.setPayload()

	// Set the Fixed header to the Packet.
	p.setFixedHeader()

	// Return the Packet.
	return p, nil
}
990764
func (em *Emitter) Subscribe(ctx context.Context, subscriber string, queryable query.Queryable, bufferSize int) (<-chan interface{}, error) {
	qry, err := queryable.Query()
	if err != nil {
		return nil, err
	}
	return em.pubsubServer.Subscribe(ctx, subscriber, qry, bufferSize)
}
643907
func (c *DefaultClient) NewSubscriber(exchange, name string, durable, autoDelete bool) Subscriber {
	return &DefaultSubscriber{
		DefaultChannelClient: DefaultChannelClient{
			ctx:      c.ctx,
			client:   c,
			exchange: exchange,
			name:     "Subscriber",
		},
		name:       name,
		durable:    durable,
		autoDelete: autoDelete,
	}
}
628157
func (s *SubscriptionProvider) Subscribe() *Subscription {
	s.Lock()
	defer s.Unlock()
	if s.subscriptions == nil {
		s.subscriptions = make(map[string]*Subscription)
	}
	sub := newSubscription(s)
	s.subscriptions[sub.id.String()] = sub

	return sub
}
599433
func NewSubscriber(topic string, h interface{}, opts ...SubscriberOption) Subscriber {
	return DefaultServer.NewSubscriber(topic, h, opts...)
}
803111
func (c *Client) Subscribe(topic string, qos packet.QOS) (SubscribeFuture, error) {
	return c.SubscribeMultiple([]packet.Subscription{
		{Topic: topic, QOS: qos},
	})
}
1013847
func (c *Local) resubscribe(subscriber string, q tmpubsub.Query) types.Subscription {
	attempts := 0
	for {
		if !c.IsRunning() {
			return nil
		}

		sub, err := c.EventBus.Subscribe(context.Background(), subscriber, q)
		if err == nil {
			return sub
		}

		attempts++
		time.Sleep((10 << uint(attempts)) * time.Millisecond) // 10ms -> 20ms -> 40ms
	}
}
544915
func (w *WebSocketService) Subscribe() error {
	// Subscribe to each channel
	if err := w.sendSubscribeMessages(); err != nil {
		return err
	}

	for {
		_, p, err := w.ws.ReadMessage()
		if err != nil {
			return err
		}

		if bytes.Contains(p, []byte("event")) {
			w.handleEventMessage(p)
		} else {
			w.handleDataMessage(p)
		}
	}
	// nolint
	return nil
}
236720
func (w *wrapper) Subscribe(c Committer) {
	w.mut.Lock()
	w.subs = append(w.subs, c)
	w.mut.Unlock()
}
655826
func (p *Pubsub) resubscribe(write *writePump) {
	timer := gaugeLatency(PromReconnectLatency)
	PromReconnections.Inc()

	p.subsMu.Lock()
	defer p.subsMu.Unlock()

	for kind, recs := range p.subs {
		if recs == nil {
			continue
		}

		for _, ev := range recs.getList() {
			write.Data() <- command{
				command: EventType(kind).SubCommand(),
				channel: ev.name,
			}
		}
	}

	timer()
}
661512
func NewSubscriptions(conn *Conn) *Subscriptions {
	s := &Subscriptions{
		channels:       make(map[string]struct{}),
		pchannels:      make(map[string]struct{}),
		conn:           conn,
		messageChannel: make(chan *Message, 100),
		ready:          true,
		readyChannel:   make(chan bool, 1),
		throwError:     false,
	}
	go s.receive()
	return s
}
863780
func SubscriberContext(ctx context.Context) SubscriberOption {
	return func(o *SubscriberOptions) {
		o.Context = ctx
	}
}
687057
func (b *pubsubBroker) Subscribe(topic string, h broker.Handler, opts ...broker.SubscribeOption) (broker.Subscriber, error) {
	options := broker.SubscribeOptions{
		AutoAck: true,
		Queue:   "q-" + uuid.New().String(),
		Context: b.options.Context,
	}

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

	ctx := context.Background()
	sub := b.client.Subscription(options.Queue)

	if createSubscription, ok := b.options.Context.Value(createSubscription{}).(bool); !ok || createSubscription {
		exists, err := sub.Exists(ctx)
		if err != nil {
			return nil, err
		}

		if !exists {
			tt := b.client.Topic(topic)
			subb, err := b.client.CreateSubscription(ctx, options.Queue, pubsub.SubscriptionConfig{
				Topic:       tt,
				AckDeadline: time.Duration(0),
			})
			if err != nil {
				return nil, err
			}
			sub = subb
		}
	}

	subscriber := &subscriber{
		options: options,
		topic:   topic,
		exit:    make(chan bool),
		sub:     sub,
	}

	go subscriber.run(h)

	return subscriber, nil
}
1330581
func (e *eventer) Subscribe() eventChannel {
	e.eventsMutex.Lock()
	defer e.eventsMutex.Unlock()
	out := make(eventChannel, eventChanBufferSize)
	e.outs[out] = out
	return out
}
1062733
func (sc *conn) Subscribe(subject string, cb MsgHandler, options ...SubscriptionOption) (Subscription, error) {
	return sc.subscribe(subject, "", cb, options...)
}
115055
func (ec *EthereumClient) SubscribeNewHead(ctx *Context, handler NewHeadHandler, buffer int) (sub *Subscription, _ error) {
	// Subscribe to the event internally
	ch := make(chan *types.Header, buffer)
	rawSub, err := ec.client.SubscribeNewHead(ctx.context, ch)
	if err != nil {
		return nil, err
	}
	// Start up a dispatcher to feed into the callback
	go func() {
		for {
			select {
			case header := <-ch:
				handler.OnNewHead(&Header{header})

			case err := <-rawSub.Err():
				if err != nil {
					handler.OnError(err.Error())
				}
				return
			}
		}
	}()
	return &Subscription{rawSub}, nil
}
57369
func (c *Client) processSubscribe(pkt *packet.Subscribe) error {
	// acquire subscribe token
	select {
	case <-c.subscribeTokens:
		// continue
	case <-time.After(c.TokenTimeout):
		return c.die(ClientError, ErrTokenTimeout)
	case <-c.tomb.Dying():
		return tomb.ErrDying
	}

	// prepare suback packet
	suback := packet.NewSuback()
	suback.ReturnCodes = make([]packet.QOS, len(pkt.Subscriptions))
	suback.ID = pkt.ID

	// set granted qos
	for i, subscription := range pkt.Subscriptions {
		suback.ReturnCodes[i] = subscription.QOS
	}

	// subscribe client to queue
	err := c.backend.Subscribe(c, pkt.Subscriptions, func() {
		select {
		case c.ackQueue <- suback:
		case <-c.tomb.Dying():
		}
	})
	if err != nil {
		return c.die(BackendError, err)
	}

	return nil
}