Method publish

nats.go:4424–4537

publish is the internal function to publish messages to a nats-server. Sends a protocol data message by queuing into the bufio writer and kicking the flush go routine. These writes should be protected.

(subj, reply string, validateReply bool, hdr, data []byte)

// Sends a protocol data message by queuing into the bufio writer
// and kicking the flush go routine. These writes should be protected.
func (nc *Conn) publish(subj, reply string, validateReply bool, hdr, data []byte) error {
	if nc == nil {
		return ErrInvalidConnection
	}
	if !nc.Opts.SkipSubjectValidation {
		if err := validateSubject(subj); err != nil {
			return err
		}
		if validateReply {
			if err := validateSubject(reply); err != nil {
				return ErrBadSubject
			}
		}
	} else if subj == _EMPTY_ {
		return ErrBadSubject
	}
	nc.mu.Lock()

	// Check if headers attempted to be sent to server that does not support them.
	if len(hdr) > 0 && !nc.info.Headers {
		nc.mu.Unlock()
		return ErrHeadersNotSupported
	}

	if nc.isClosed() {
		nc.mu.Unlock()
		return ErrConnectionClosed
	}

	if nc.isDrainingPubs() {
		nc.mu.Unlock()
		return ErrConnectionDraining
	}

	// Proactively reject payloads over the threshold set by server.
	msgSize := int64(len(data) + len(hdr))
	// Skip this check if we are not yet connected (RetryOnFailedConnect)
	if !nc.initc && msgSize > nc.info.MaxPayload {
		nc.mu.Unlock()
		return ErrMaxPayload
	}

	// Check if we are reconnecting, and if so check if
	// we have exceeded our reconnect outbound buffer limits.
	if nc.bw.atLimitIfUsingPending() {
		nc.mu.Unlock()
		return ErrReconnectBufExceeded
	}

	var mh []byte
	if hdr != nil {
		mh = nc.scratch[:len(_HPUB_P_)]
	} else {
		mh = nc.scratch[1:len(_HPUB_P_)]
	}
	mh = append(mh, subj...)
	mh = append(mh, ' ')
	if reply != "" {