Method newAsyncReply

jetstream/publish.go:427–468

()

)

func (js *jetStream) newAsyncReply() (string, error) {
	js.publisher.Lock()
	if js.publisher.replySub == nil {
		// Create our wildcard reply subject.
		sha := sha256.New()
		sha.Write([]byte(nuid.Next()))
		b := sha.Sum(nil)
		for i := 0; i < aReplyTokensize; i++ {
			b[i] = rdigits[int(b[i]%base)]
		}
		js.publisher.replyPrefix = fmt.Sprintf("%s%s.", js.opts.replyPrefix, b[:aReplyTokensize])
		sub, err := js.conn.Subscribe(fmt.Sprintf("%s*", js.publisher.replyPrefix), js.handleAsyncReply)
		if err != nil {
			js.publisher.Unlock()
			return "", err
		}
		js.publisher.replySub = sub
		js.publisher.rr = rand.New(rand.NewSource(time.Now().UnixNano()))
	}
	if js.publisher.connStatusCh == nil {
		js.publisher.connStatusCh = js.conn.StatusChanged(nats.RECONNECTING, nats.CLOSED)
		go js.resetPendingAcksOnReconnect()
	}
	var sb strings.Builder
	sb.WriteString(js.publisher.replyPrefix)
	for {
		rn := js.publisher.rr.Int63()
		var b [aReplyTokensize]byte
		for i, l := 0, rn; i < len(b); i++ {
			b[i] = rdigits[l%base]
			l /= base
		}
		if _, ok := js.publisher.acks[string(b[:])]; ok {
			continue
		}
		sb.Write(b[:])
		break
	}

	js.publisher.Unlock()
	return sb.String(), nil
}

// Handle an async reply from PublishAsync.
func (js *jetStream) handleAsyncReply(m *nats.Msg) {