Function TestJetStreamRePublish

test/js_test.go:9806–9871

(t *testing.T)

}

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

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

	if _, err := js.AddStream(&nats.StreamConfig{
		Name:     "RP",
		Storage:  nats.MemoryStorage,
		Subjects: []string{"foo", "bar", "baz"},
		RePublish: &nats.RePublish{
			Source:      ">",
			Destination: "RP.>",
		},
	}); err != nil {
		t.Fatalf("Error adding stream: %v", err)
	}

	sub, err := nc.SubscribeSync("RP.>")
	if err != nil {
		t.Fatalf("Error on subscribe: %v", err)
	}

	msg, toSend := []byte("OK TO REPUBLISH?"), 100
	for i := 0; i < toSend; i++ {
		js.Publish("foo", msg)
		js.Publish("bar", msg)
		js.Publish("baz", msg)
	}

	lseq := map[string]int{
		"foo": 0,
		"bar": 0,
		"baz": 0,
	}

	for i := 1; i <= toSend; i++ {
		m, err := sub.NextMsg(time.Second)
		if err != nil {
			t.Fatalf("Error on next msg: %v", err)
		}
		// Grab info from Header
		stream := m.Header.Get(nats.JSStream)
		if stream != "RP" {
			t.Fatalf("Unexpected header: %+v", m.Header)
		}
		// Make sure sequence is correct.
		seq, err := strconv.Atoi(m.Header.Get(nats.JSSequence))
		if err != nil {
			t.Fatalf("Error decoding sequence for %s", m.Header.Get(nats.JSSequence))
		}
		if seq != i {
			t.Fatalf("Expected sequence to be %v, got %v", i, seq)
		}
		// Make sure last sequence matches last seq we received on this subject.
		last, err := strconv.Atoi(m.Header.Get(nats.JSLastSequence))
		if err != nil {