[ADDED] Options() and Conn() methods to JetStream (#1792)

Signed-off-by: Piotr Piotrowski <[email protected]>
nats-io · Feb 4, 2025 · faec055 · faec055
1 parent 16e780e
commit faec055
Show file tree

Hide file tree

Showing 11 changed files with 220 additions and 152 deletions.
diff --git a/jetstream/api.go b/jetstream/api.go
@@ -116,12 +116,13 @@ func (js *jetStream) apiRequestJSON(ctx context.Context, subject string, resp an
 
 // a RequestWithContext with tracing via TraceCB
 func (js *jetStream) apiRequest(ctx context.Context, subj string, data ...[]byte) (*jetStreamMsg, error) {
+	subj = js.apiSubject(subj)
 	var req []byte
 	if len(data) > 0 {
 		req = data[0]
 	}
-	if js.clientTrace != nil {
-		ctrace := js.clientTrace
+	if js.opts.clientTrace != nil {
+		ctrace := js.opts.clientTrace
 		if ctrace.RequestSent != nil {
 			ctrace.RequestSent(subj, req)
 		}
@@ -130,8 +131,8 @@ func (js *jetStream) apiRequest(ctx context.Context, subj string, data ...[]byte
 	if err != nil {
 		return nil, err
 	}
-	if js.clientTrace != nil {
-		ctrace := js.clientTrace
+	if js.opts.clientTrace != nil {
+		ctrace := js.opts.clientTrace
 		if ctrace.ResponseReceived != nil {
 			ctrace.ResponseReceived(subj, resp.Data, resp.Header)
 		}
@@ -140,12 +141,12 @@ func (js *jetStream) apiRequest(ctx context.Context, subj string, data ...[]byte
 	return js.toJSMsg(resp), nil
 }
 
-func apiSubj(prefix, subject string) string {
-	if prefix == "" {
-		return subject
+func (js *jetStream) apiSubject(subj string) string {
+	if js.opts.apiPrefix == "" {
+		return subj
 	}
 	var b strings.Builder
-	b.WriteString(prefix)
-	b.WriteString(subject)
+	b.WriteString(js.opts.apiPrefix)
+	b.WriteString(subj)
 	return b.String()
 }
diff --git a/jetstream/consumer.go b/jetstream/consumer.go
@@ -155,14 +155,14 @@ type (
 
 // Info fetches current ConsumerInfo from the server.
 func (p *pullConsumer) Info(ctx context.Context) (*ConsumerInfo, error) {
-	ctx, cancel := wrapContextWithoutDeadline(ctx)
+	ctx, cancel := p.js.wrapContextWithoutDeadline(ctx)
 	if cancel != nil {
 		defer cancel()
 	}
-	infoSubject := apiSubj(p.jetStream.apiPrefix, fmt.Sprintf(apiConsumerInfoT, p.stream, p.name))
+	infoSubject := fmt.Sprintf(apiConsumerInfoT, p.stream, p.name)
 	var resp consumerInfoResponse
 
-	if _, err := p.jetStream.apiRequestJSON(ctx, infoSubject, &resp); err != nil {
+	if _, err := p.js.apiRequestJSON(ctx, infoSubject, &resp); err != nil {
 		return nil, err
 	}
 	if resp.Error != nil {
@@ -187,7 +187,7 @@ func (p *pullConsumer) CachedInfo() *ConsumerInfo {
 }
 
 func upsertConsumer(ctx context.Context, js *jetStream, stream string, cfg ConsumerConfig, action string) (Consumer, error) {
-	ctx, cancel := wrapContextWithoutDeadline(ctx)
+	ctx, cancel := js.wrapContextWithoutDeadline(ctx)
 	if cancel != nil {
 		defer cancel()
 	}
@@ -218,9 +218,9 @@ func upsertConsumer(ctx context.Context, js *jetStream, stream string, cfg Consu
 		if err := validateSubject(cfg.FilterSubject); err != nil {
 			return nil, err
 		}
-		ccSubj = apiSubj(js.apiPrefix, fmt.Sprintf(apiConsumerCreateWithFilterSubjectT, stream, consumerName, cfg.FilterSubject))
+		ccSubj = fmt.Sprintf(apiConsumerCreateWithFilterSubjectT, stream, consumerName, cfg.FilterSubject)
 	} else {
-		ccSubj = apiSubj(js.apiPrefix, fmt.Sprintf(apiConsumerCreateT, stream, consumerName))
+		ccSubj = fmt.Sprintf(apiConsumerCreateT, stream, consumerName)
 	}
 	var resp consumerInfoResponse
 
@@ -240,12 +240,12 @@ func upsertConsumer(ctx context.Context, js *jetStream, stream string, cfg Consu
 	}
 
 	return &pullConsumer{
-		jetStream: js,
-		stream:    stream,
-		name:      resp.Name,
-		durable:   cfg.Durable != "",
-		info:      resp.ConsumerInfo,
-		subs:      syncx.Map[string, *pullSubscription]{},
+		js:      js,
+		stream:  stream,
+		name:    resp.Name,
+		durable: cfg.Durable != "",
+		info:    resp.ConsumerInfo,
+		subs:    syncx.Map[string, *pullSubscription]{},
 	}, nil
 }
 
@@ -267,14 +267,14 @@ func generateConsName() string {
 }
 
 func getConsumer(ctx context.Context, js *jetStream, stream, name string) (Consumer, error) {
-	ctx, cancel := wrapContextWithoutDeadline(ctx)
+	ctx, cancel := js.wrapContextWithoutDeadline(ctx)
 	if cancel != nil {
 		defer cancel()
 	}
 	if err := validateConsumerName(name); err != nil {
 		return nil, err
 	}
-	infoSubject := apiSubj(js.apiPrefix, fmt.Sprintf(apiConsumerInfoT, stream, name))
+	infoSubject := fmt.Sprintf(apiConsumerInfoT, stream, name)
 
 	var resp consumerInfoResponse
 
@@ -292,26 +292,26 @@ func getConsumer(ctx context.Context, js *jetStream, stream, name string) (Consu
 	}
 
 	cons := &pullConsumer{
-		jetStream: js,
-		stream:    stream,
-		name:      name,
-		durable:   resp.Config.Durable != "",
-		info:      resp.ConsumerInfo,
-		subs:      syncx.Map[string, *pullSubscription]{},
+		js:      js,
+		stream:  stream,
+		name:    name,
+		durable: resp.Config.Durable != "",
+		info:    resp.ConsumerInfo,
+		subs:    syncx.Map[string, *pullSubscription]{},
 	}
 
 	return cons, nil
 }
 
 func deleteConsumer(ctx context.Context, js *jetStream, stream, consumer string) error {
-	ctx, cancel := wrapContextWithoutDeadline(ctx)
+	ctx, cancel := js.wrapContextWithoutDeadline(ctx)
 	if cancel != nil {
 		defer cancel()
 	}
 	if err := validateConsumerName(consumer); err != nil {
 		return err
 	}
-	deleteSubject := apiSubj(js.apiPrefix, fmt.Sprintf(apiConsumerDeleteT, stream, consumer))
+	deleteSubject := fmt.Sprintf(apiConsumerDeleteT, stream, consumer)
 
 	var resp consumerDeleteResponse