Function TestJetStreamCrossAccountMirrorsAndSources

test/js_test.go:4550–4682

(t *testing.T)

}

func TestJetStreamCrossAccountMirrorsAndSources(t *testing.T) {
	conf := createConfFile(t, []byte(`
		listen: 127.0.0.1:-1
		no_auth_user: rip
		jetstream: {max_mem_store: 64GB, max_file_store: 10TB}
		accounts {
			JS {
				jetstream: enabled
				users = [ { user: "rip", pass: "pass" } ]
				exports [
					{ service: "$JS.API.CONSUMER.>" } # To create internal consumers to mirror/source.
					{ stream: "RI.DELIVER.SYNC.>" }   # For the mirror/source consumers sending to IA via delivery subject.
				]
			}
			IA {
				jetstream: enabled
				users = [ { user: "dlc", pass: "pass" } ]
				imports [
					{ service: { account: JS, subject: "$JS.API.CONSUMER.>"}, to: "RI.JS.API.CONSUMER.>" }
					{ stream: { account: JS, subject: "RI.DELIVER.SYNC.>"} }
				]
			}
			$SYS { users = [ { user: "admin", pass: "s3cr3t!" } ] }
		}
	`))
	defer os.Remove(conf)

	s, _ := RunServerWithConfig(conf)
	defer shutdownJSServerAndRemoveStorage(t, s)

	nc1, js1 := jsClient(t, s, nats.UserInfo("rip", "pass"))
	defer nc1.Close()

	var err error

	_, err = js1.AddStream(&nats.StreamConfig{
		Name:     "TEST",
		Replicas: 1,
	})
	if err != nil {
		t.Fatal(err)
	}

	const (
		toSend      = 100
		publishSubj = "TEST"
		sourceName  = "MY_SOURCE_TEST"
		mirrorName  = "MY_MIRROR_TEST"
	)
	for i := 0; i < toSend; i++ {
		data := []byte(fmt.Sprintf("OK %d", i))
		if _, err := js1.Publish(publishSubj, data); err != nil {
			t.Fatalf("Unexpected publish error: %v", err)
		}
	}

	nc2, js2 := jsClient(t, s, nats.UserInfo("dlc", "pass"))
	defer nc2.Close()