Function checkMsg

jetstream/message.go:461–506

checkMsg returns whether the given message is a user message or a control message. If the status header is present, it returns an appropriate error based on the status code (404, etc.)

(msg *nats.Msg)

// If the status header is present, it returns an appropriate error based
// on the status code (404, etc.)
func checkMsg(msg *nats.Msg) (bool, error) {
	// If payload or no header, consider this a user message
	if len(msg.Data) > 0 || len(msg.Header) == 0 {
		return true, nil
	}
	// Look for status header
	val := msg.Header.Get("Status")
	descr := msg.Header.Get("Description")
	// If not present, then this is considered a user message
	if val == "" {
		return true, nil
	}

	switch val {
	case statusBadRequest:
		return false, ErrBadRequest
	case statusNoResponders:
		return false, nats.ErrNoResponders
	case statusNoMsgs:
		// 404 indicates that there are no messages.
		return false, ErrNoMessages
	case statusTimeout:
		return false, nats.ErrTimeout
	case statusControlMsg:
		return false, nil
	case statusPinIdMismatch:
		return false, ErrPinIDMismatch
	case statusConflict:
		if strings.Contains(strings.ToLower(descr), maxBytesExceeded) {
			return false, ErrMaxBytesExceeded
		}
		if strings.Contains(strings.ToLower(descr), batchCompleted) {
			return false, ErrBatchCompleted
		}
		if strings.Contains(strings.ToLower(descr), consumerDeleted) {
			return false, ErrConsumerDeleted
		}
		if strings.Contains(strings.ToLower(descr), leadershipChange) {
			return false, ErrConsumerLeadershipChanged
		}
		if strings.Contains(strings.ToLower(descr), serverShutdown) {
			return false, ErrServerShutdown
		}
	}
	return false, fmt.Errorf("nats: %s", msg.Header.Get("Description"))
}

func parsePending(msg *nats.Msg) (int, int, error) {
	msgsLeftStr := msg.Header.Get("Nats-Pending-Messages")