Method ackReply

js.go:3577–3652

ackReply handles all acks. Will do the right thing for pull and sync mode. It ensures that an ack is only sent a single time, regardless of how many times it is being called to avoid duplicated acks.

(ackType []byte, sync bool, opts ...AckOpt)

// It ensures that an ack is only sent a single time, regardless of
// how many times it is being called to avoid duplicated acks.
func (m *Msg) ackReply(ackType []byte, sync bool, opts ...AckOpt) error {
	var o ackOpts
	for _, opt := range opts {
		if err := opt.configureAck(&o); err != nil {
			return err
		}
	}

	if err := m.checkReply(); err != nil {
		return err
	}

	var ackNone bool
	var js *js

	sub := m.Sub
	sub.mu.Lock()
	nc := sub.conn
	if jsi := sub.jsi; jsi != nil {
		js = jsi.js
		ackNone = jsi.ackNone
	}
	sub.mu.Unlock()

	// Skip if already acked.
	if atomic.LoadUint32(&m.ackd) == 1 {
		return ErrMsgAlreadyAckd
	}
	if ackNone {
		return ErrCantAckIfConsumerAckNone
	}

	usesCtx := o.ctx != nil
	usesWait := o.ttl > 0

	// Only allow either AckWait or Context option to set the timeout.
	if usesWait && usesCtx {
		return ErrContextAndTimeout
	}

	sync = sync || usesCtx || usesWait
	ctx := o.ctx
	wait := defaultRequestWait
	if usesWait {
		wait = o.ttl
	} else if js != nil {
		wait = js.opts.wait
	}

	var body []byte
	var err error
	// This will be > 0 only when called from NakWithDelay()
	if o.nakDelay > 0 {
		body = []byte(fmt.Sprintf("%s {\"delay\": %d}", ackType, o.nakDelay.Nanoseconds()))
	} else {
		body = ackType
	}