Method PublishMsg

jetstream/publish.go:168–262

PublishMsg performs a synchronous publish to a stream and waits for ack from server. It accepts subject name (which must be bound to a stream) and nats.Message.

(ctx context.Context, m *nats.Msg, opts ...PublishOpt)

// ack from server. It accepts subject name (which must be bound to a
// stream) and nats.Message.
func (js *jetStream) PublishMsg(ctx context.Context, m *nats.Msg, opts ...PublishOpt) (*PubAck, error) {
	ctx, cancel := js.wrapContextWithoutDeadline(ctx)
	if cancel != nil {
		defer cancel()
	}
	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
			}
		}
	}
	if o.stallWait > 0 {
		return nil, fmt.Errorf("%w: stall wait cannot be set to sync publish", ErrInvalidOption)
	}

	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 != "" {