Function addEndpoint

micro/service.go:418–484

(s *service, name, subject string, handler Handler, metadata map[string]string, queueGroup string, noQueue bool, msgLimit, bytesLimit int)

}

func addEndpoint(s *service, name, subject string, handler Handler, metadata map[string]string, queueGroup string, noQueue bool, msgLimit, bytesLimit int) error {
	if !nameRegexp.MatchString(name) {
		return fmt.Errorf("%w: invalid endpoint name", ErrConfigValidation)
	}
	if !subjectRegexp.MatchString(subject) {
		return fmt.Errorf("%w: invalid endpoint subject", ErrConfigValidation)
	}
	if !subjectRegexp.MatchString(queueGroup) {
		return fmt.Errorf("%w: invalid endpoint queue group", ErrConfigValidation)
	}
	endpoint := &Endpoint{
		service: s,
		EndpointConfig: EndpointConfig{
			Subject:            subject,
			Handler:            handler,
			Metadata:           metadata,
			QueueGroup:         queueGroup,
			QueueGroupDisabled: noQueue,
		},
		Name: name,
	}

	var sub *nats.Subscription
	var err error
	var options = endpointOpts{
		msgLimit:   msgLimit,
		bytesLimit: bytesLimit,
	}

	if !noQueue {
		sub, err = s.nc.QueueSubscribe(
			subject,
			queueGroup,
			func(m *nats.Msg) {
				s.reqHandler(endpoint, &request{msg: m})
			},
		)
	} else {
		sub, err = s.nc.Subscribe(
			subject,
			func(m *nats.Msg) {
				s.reqHandler(endpoint, &request{msg: m})
			},
		)
	}
	if err != nil {
		return err
	}

	// Apply pending limits if configured
	if options.msgLimit != 0 || options.bytesLimit != 0 {
		if err := sub.SetPendingLimits(options.msgLimit, options.bytesLimit); err != nil {
			return err
		}
	}

	s.m.Lock()
	endpoint.subscription = sub