Function upsertConsumer

jetstream/consumer.go:287–349

(ctx context.Context, js *jetStream, stream string, cfg ConsumerConfig, action string)

}

func upsertConsumer(ctx context.Context, js *jetStream, stream string, cfg ConsumerConfig, action string) (*consumerInfoResponse, error) {
	ctx, cancel := js.wrapContextWithoutDeadline(ctx)
	if cancel != nil {
		defer cancel()
	}
	req := createConsumerRequest{
		Stream: stream,
		Config: &cfg,
		Action: action,
	}
	reqJSON, err := json.Marshal(req)
	if err != nil {
		return nil, err
	}

	consumerName := cfg.Name
	if consumerName == "" {
		if cfg.Durable != "" {
			consumerName = cfg.Durable
		} else {
			consumerName = generateConsName()
		}
	}
	if err := validateConsumerName(consumerName); err != nil {
		return nil, err
	}

	var ccSubj string
	if cfg.FilterSubject != "" && len(cfg.FilterSubjects) == 0 {
		if err := validateSubject(cfg.FilterSubject); err != nil {
			return nil, err
		}
		ccSubj = fmt.Sprintf(apiConsumerCreateWithFilterSubjectT, stream, consumerName, cfg.FilterSubject)
	} else {
		ccSubj = fmt.Sprintf(apiConsumerCreateT, stream, consumerName)
	}
	var resp consumerInfoResponse

	if _, err := js.apiRequestJSON(ctx, ccSubj, &resp, reqJSON); err != nil {
		return nil, err
	}
	if resp.Error != nil {
		if resp.Error.ErrorCode == JSErrCodeStreamNotFound {
			return nil, ErrStreamNotFound
		}
		if resp.Error.ErrorCode == JSErrCodeMaximumConsumersLimit {
			return nil, ErrMaximumConsumersLimit
		}

		return nil, resp.Error
	}

	if resp.Error == nil && resp.ConsumerInfo == nil {
		return nil, ErrConsumerCreationResponseEmpty
	}

	// check whether multiple filter subjects (if used) are reflected in the returned ConsumerInfo
	if len(cfg.FilterSubjects) != 0 && len(resp.Config.FilterSubjects) == 0 {