Function TestObjectStoreMirror

test/object_test.go:1149–1215

(t *testing.T)

}

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),
			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)

	// Make sure we sync.
	checkFor(t, 2*time.Second, 15*time.Millisecond, func() error {
		mirrorValue, err := mirrorObs.GetString("A")
		if err != nil {
			return err
		}
		if mirrorValue != "abc" {
			t.Fatalf("Expected mirrored object store value to be the same as original")
		}
		return nil
	})

	watcher, err := mirrorObs.Watch()
	if err != nil {
		t.Fatalf("Error creating watcher: %v", err)
	}
	defer watcher.Stop()

	// expect to get one value and nil
	for {
		select {