Method subscribe

js.go:1673–2058

(subj, queue string, cb MsgHandler, ch chan *Msg, isSync, isPullMode bool, opts []SubOpt)

}

func (js *js) subscribe(subj, queue string, cb MsgHandler, ch chan *Msg, isSync, isPullMode bool, opts []SubOpt) (*Subscription, error) {
	cfg := ConsumerConfig{
		DeliverPolicy: deliverPolicyNotSet,
		AckPolicy:     ackPolicyNotSet,
		ReplayPolicy:  replayPolicyNotSet,
	}
	o := subOpts{cfg: &cfg}
	if len(opts) > 0 {
		for _, opt := range opts {
			if opt == nil {
				continue
			}
			if err := opt.configureSubscribe(&o); err != nil {
				return nil, err
			}
		}
	}

	// If no stream name is specified, the subject cannot be empty.
	if subj == _EMPTY_ && o.stream == _EMPTY_ {
		return nil, errors.New("nats: subject required")
	}

	// Note that these may change based on the consumer info response we may get.
	hasHeartbeats := o.cfg.Heartbeat > 0
	hasFC := o.cfg.FlowControl

	// Some checks for pull subscribers
	if isPullMode {
		// No deliver subject should be provided
		if o.cfg.DeliverSubject != _EMPTY_ {
			return nil, ErrPullSubscribeToPushConsumer
		}
	}

	// Some check/setting specific to queue subs
	if queue != _EMPTY_ {
		// Queue subscriber cannot have HB or FC (since messages will be randomly dispatched
		// to members). We may in the future have a separate NATS subscription that all members
		// would subscribe to and server would send on.
		if o.cfg.Heartbeat > 0 || o.cfg.FlowControl {
			// Not making this a public ErrXXX in case we allow in the future.
			return nil, errors.New("nats: queue subscription doesn't support idle heartbeat nor flow control")
		}

		// If this is a queue subscription and no consumer nor durable name was specified,
		// then we will use the queue name as a durable name.
		if o.consumer == _EMPTY_ && o.cfg.Durable == _EMPTY_ {
			if err := checkConsumerName(queue); err != nil {
				return nil, err
			}
			o.cfg.Durable = queue
		}
	}

	var (
		err           error
		shouldCreate  bool