Method PublishMsgAsync

jetstream/publish.go:274–420

PublishMsgAsync performs an asynchronous publish to a stream and returns [PubAckFuture] interface. It accepts subject name (which must be bound to a stream) and nats.Message.

(m *nats.Msg, opts ...PublishOpt)

// returns [PubAckFuture] interface. It accepts subject name (which must
// be bound to a stream) and nats.Message.
func (js *jetStream) PublishMsgAsync(m *nats.Msg, opts ...PublishOpt) (PubAckFuture, error) {
	o := pubOpts{
		retryWait:     DefaultPubRetryWait,
		retryAttempts: DefaultPubRetryAttempts,
	}
	if len(opts) > 0 {
		if m.Header == nil {
			m.Header = nats.Header{}
		}
		for _, opt := range opts {
			if err := opt(&o); err != nil {
				return nil, err
			}
		}
	}
	defaultStallWait := 200 * time.Millisecond

	stallWait := defaultStallWait
	if o.stallWait > 0 {
		stallWait = o.stallWait
	}

	if o.id != "" {
		m.Header.Set(MsgIDHeader, o.id)
	}
	if o.lastMsgID != "" {
		m.Header.Set(ExpectedLastMsgIDHeader, o.lastMsgID)
	}
	if o.stream != "" {
		m.Header.Set(ExpectedStreamHeader, o.stream)
	}
	if o.lastSeq != nil {
		m.Header.Set(ExpectedLastSeqHeader, strconv.FormatUint(*o.lastSeq, 10))
	}
	if o.lastSubjectSeq != nil {
		m.Header.Set(ExpectedLastSubjSeqHeader, strconv.FormatUint(*o.lastSubjectSeq, 10))
	}
	if o.lastSubject != "" {
		m.Header.Set(ExpectedLastSubjSeqSubjHeader, o.lastSubject)
		m.Header.Set(ExpectedLastSubjSeqHeader, strconv.FormatUint(*o.lastSubjectSeq, 10))
	}
	if o.ttl > 0 {
		m.Header.Set(MsgTTLHeader, o.ttl.String())
	}
	if o.schedule != "" {
		m.Header.Set(ScheduleHeader, o.schedule)
	}
	if o.scheduleTarget != "" {
		m.Header.Set(ScheduleTargetHeader, o.scheduleTarget)
	}
	if o.scheduleSource != "" {
		m.Header.Set(ScheduleSourceHeader, o.scheduleSource)
	}
	if o.scheduleTTL != "" {
		m.Header.Set(ScheduleTTLHeader, o.scheduleTTL)
	}
	if o.scheduleTZ != "" {
		m.Header.Set(ScheduleTimeZoneHeader, o.scheduleTZ)