Method ackReply

jetstream/message.go:399–446

(ctx context.Context, ackType ackType, sync bool, opts ackOpts)

}

func (m *jetStreamMsg) ackReply(ctx context.Context, ackType ackType, sync bool, opts ackOpts) error {
	err := m.checkReply()
	if err != nil {
		return err
	}

	m.Lock()
	if m.ackd {
		m.Unlock()
		return ErrMsgAlreadyAckd
	}
	m.Unlock()

	if sync {
		var cancel context.CancelFunc
		ctx, cancel = m.js.wrapContextWithoutDeadline(ctx)
		if cancel != nil {
			defer cancel()
		}
	}

	var body []byte
	if opts.nakDelay > 0 {
		body = []byte(fmt.Sprintf("%s {\"delay\": %d}", ackType, opts.nakDelay.Nanoseconds()))
	} else if opts.termReason != "" {
		body = []byte(fmt.Sprintf("%s %s", ackType, opts.termReason))
	} else {
		body = ackType
	}

	if sync {
		_, err = m.js.conn.RequestWithContext(ctx, m.msg.Reply, body)
	} else {
		err = m.js.conn.Publish(m.msg.Reply, body)
	}
	if err != nil {
		return err
	}

	// Mark that the message has been acked unless it is ackProgress
	// which can be sent many times.
	if !bytes.Equal(ackType, ackProgress) {
		m.Lock()
		m.ackd = true
		m.Unlock()
	}
	return nil
}

func (m *jetStreamMsg) checkReply() error {
	if m == nil || m.msg.Sub == nil {