chore: Bind Object Store bucket stream when getting object.

The change is needed to properly operate on mirrored object store buckets.
By binding stream nats.go does not need to search for stream by subjects (which does not work for mirrors).

jetstream/object.go

@@ -29,9 +29,10 @@ import (
 	"sync"
 	"time"
 
+	"github.com/nats-io/nuid"
+
 	"github.com/nats-io/nats.go"
 	"github.com/nats-io/nats.go/internal/parser"
-	"github.com/nats-io/nuid"
 )
 
 type (
@@ -918,7 +919,13 @@ func (obs *obs) Get(ctx context.Context, name string, opts ...GetObjectOpt) (Obj
 	}
 
 	chunkSubj := fmt.Sprintf(objChunksPreTmpl, obs.name, info.NUID)
-	_, err = obs.pushJS.Subscribe(chunkSubj, processChunk, nats.OrderedConsumer(), nats.Context(ctx))
+	streamName := fmt.Sprintf(objNameTmpl, obs.name)
+	subscribeOpts := []nats.SubOpt{
+		nats.OrderedConsumer(),
+		nats.Context(ctx),
+		nats.BindStream(streamName),
+	}
+	_, err = obs.pushJS.Subscribe(chunkSubj, processChunk, subscribeOpts...)
 	if err != nil {
 		return nil, err
 	}

object.go

@@ -29,8 +29,9 @@ import (
 	"sync"
 	"time"
 
-	"github.com/nats-io/nats.go/internal/parser"
 	"github.com/nats-io/nuid"
+
+	"github.com/nats-io/nats.go/internal/parser"
 )
 
 // ObjectStoreManager creates, loads and deletes Object Stores
@@ -694,7 +695,12 @@ func (obs *obs) Get(name string, opts ...GetObjectOpt) (ObjectResult, error) {
 	}
 
 	chunkSubj := fmt.Sprintf(objChunksPreTmpl, obs.name, info.NUID)
-	_, err = obs.js.Subscribe(chunkSubj, processChunk, OrderedConsumer())
+	streamName := fmt.Sprintf(objNameTmpl, obs.name)
+	subscribeOpts := []SubOpt{
+		OrderedConsumer(),
+		BindStream(streamName),
+	}
+	_, err = obs.js.Subscribe(chunkSubj, processChunk, subscribeOpts...)
 	if err != nil {
 		return nil, err
 	}

test/object_test.go

@@ -1121,3 +1121,51 @@ func TestObjectStoreCompression(t *testing.T) {
 		t.Fatalf("Expected stream to be compressed with S2")
 	}
 }
+
+func TestObjectStoreMirror(t *testing.T) {
+	s := RunBasicJetStreamServer()
+	defer shutdownJSServerAndRemoveStorage(t, s)
+
+	nc, js := jsClient(t, s)
+	defer nc.Close()
+
+	bucketName := "test-bucket"
+
+	obs, err := js.CreateObjectStore(&nats.ObjectStoreConfig{Bucket: bucketName, Description: "testing"})
+	expectOk(t, err)
+
+	mirrorBucketName := "mirror-test-bucket"
+
+	_, err = js.AddStream(&nats.StreamConfig{
+		Name: fmt.Sprintf("OBJ_%s", mirrorBucketName),
+		Mirror: &nats.StreamSource{
+			Name: fmt.Sprintf("OBJ_%s", bucketName),
+			//FilterSubject: fmt.Sprintf("$O.%s.>", bucketName),
+			SubjectTransforms: []nats.SubjectTransformConfig{
+				{
+					Source:      fmt.Sprintf("$O.%s.>", bucketName),
+					Destination: fmt.Sprintf("$O.%s.>", mirrorBucketName),
+				},
+			},
+		},
+		AllowRollup: true, // meta messages are always rollups
+	})
+	if err != nil {
+		t.Fatalf("Error creating object store bucket mirror: %v", err)
+	}
+
+	_, err = obs.PutString("A", "abc")
+	expectOk(t, err)
+
+	mirrorObs, err := js.ObjectStore(mirrorBucketName)
+	expectOk(t, err)
+
+	time.Sleep(time.Second * 2)
+
+	mirrorValue, err := mirrorObs.GetString("A")
+	expectOk(t, err)
+
+	if mirrorValue != "abc" {
+		t.Fatalf("Expected mirrored object store value to be the same as original")
+	}
+}