Method setDefaults

jetstream/pull.go:1166–1212

(ordered bool)

}

func (consumeOpts *consumeOpts) setDefaults(ordered bool) error {
	// we cannot use both max messages and max bytes unless we're using max bytes as fetch size limiter
	if consumeOpts.MaxBytes != unset && consumeOpts.MaxMessages != unset && !consumeOpts.LimitSize {
		return errors.New("only one of MaxMessages and MaxBytes can be specified")
	}
	if consumeOpts.MaxBytes != unset && !consumeOpts.LimitSize {
		// we used PullMaxBytes setting, set MaxMessages to a high value
		consumeOpts.MaxMessages = defaultBatchMaxBytesOnly
	} else if consumeOpts.MaxMessages == unset {
		// otherwise, if max messages is not set, set it to default value
		consumeOpts.MaxMessages = DefaultMaxMessages
	}
	// if user did not set max bytes, set it to 0
	if consumeOpts.MaxBytes == unset {
		consumeOpts.MaxBytes = 0
	}

	if consumeOpts.ThresholdMessages == 0 {
		// half of the max messages, rounded up
		consumeOpts.ThresholdMessages = int(math.Ceil(float64(consumeOpts.MaxMessages) / 2))
	}
	if consumeOpts.ThresholdBytes == 0 {
		// half of the max bytes, rounded up
		consumeOpts.ThresholdBytes = int(math.Ceil(float64(consumeOpts.MaxBytes) / 2))
	}

	// set default heartbeats
	if consumeOpts.Heartbeat == unset {
		// by default, use 50% of expiry time
		consumeOpts.Heartbeat = consumeOpts.Expires / 2
		if ordered {
			// for ordered consumers, the default heartbeat is 5 seconds
			if consumeOpts.Expires < 10*time.Second {
				consumeOpts.Heartbeat = consumeOpts.Expires / 2
			} else {
				consumeOpts.Heartbeat = 5 * time.Second
			}
		} else if consumeOpts.Heartbeat > 30*time.Second {
			// cap the heartbeat to 30 seconds
			consumeOpts.Heartbeat = 30 * time.Second
		}
	}
	if consumeOpts.Heartbeat > consumeOpts.Expires/2 {
		return fmt.Errorf("%w: the value of Heartbeat must be less than 50%% of expiry", ErrInvalidOption)
	}
	return nil
}

func (c *pullConsumer) getPinID() string {
	c.Lock()