Function TestKeyValueSourcing

test/kv_test.go:1675–1735

(t *testing.T)

}

func TestKeyValueSourcing(t *testing.T) {
	s := RunBasicJetStreamServer()
	defer shutdownJSServerAndRemoveStorage(t, s)

	nc, js := jsClient(t, s)
	defer nc.Close()

	kvA, err := js.CreateKeyValue(&nats.KeyValueConfig{Bucket: "A"})
	if err != nil {
		t.Fatalf("Error creating kv: %v", err)
	}

	_, err = kvA.Create("keyA", []byte("1"))
	if err != nil {
		t.Fatalf("Error creating key: %v", err)
	}

	if _, err := kvA.Get("keyA"); err != nil {
		t.Fatalf("Got error getting keyA from A: %v", err)
	}

	kvB, err := js.CreateKeyValue(&nats.KeyValueConfig{Bucket: "B"})
	if err != nil {
		t.Fatalf("Error creating kv: %v", err)
	}

	_, err = kvB.Create("keyB", []byte("1"))
	if err != nil {
		t.Fatalf("Error creating key: %v", err)
	}

	kvC, err := js.CreateKeyValue(&nats.KeyValueConfig{Bucket: "C", Sources: []*nats.StreamSource{{Name: "A"}, {Name: "B"}}})
	if err != nil {
		t.Fatalf("Error creating kv: %v", err)
	}

	i := 0
	for {
		status, err := kvC.Status()
		if err != nil {
			t.Fatalf("Error getting bucket status: %v", err)
		}
		if status.Values() == 2 {
			break
		} else {
			i++
			if i > 10 {
				t.Fatalf("Error sourcing bucket does not contain the expected number of values")
			}
		}
		time.Sleep(100 * time.Millisecond)
	}

	if _, err := kvC.Get("keyA"); err != nil {
		t.Fatalf("Got error getting keyA from C: %v", err)
	}

	if _, err := kvC.Get("keyB"); err != nil {