Function TestJetStreamMsgSubjectRewrite

test/js_test.go:9269–9308

(t *testing.T)

}

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

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

	if _, err := js.AddStream(&nats.StreamConfig{
		Name:     "TEST",
		Subjects: []string{"foo"},
	}); err != nil {
		t.Fatalf("Error adding stream: %v", err)
	}

	sub, err := nc.SubscribeSync(nats.NewInbox())
	if err != nil {
		t.Fatalf("Error on subscribe: %v", err)
	}
	if _, err := js.AddConsumer("TEST", &nats.ConsumerConfig{
		DeliverSubject: sub.Subject,
		DeliverPolicy:  nats.DeliverAllPolicy,
	}); err != nil {
		t.Fatalf("Error adding consumer: %v", err)
	}

	if _, err := js.Publish("foo", []byte("msg")); err != nil {
		t.Fatalf("Error on publish: %v", err)
	}

	msg, err := sub.NextMsg(time.Second)
	if err != nil {
		t.Fatalf("Did not get message: %v", err)
	}
	if msg.Subject != "foo" {
		t.Fatalf("Subject should be %q, got %q", "foo", msg.Subject)
	}
	if string(msg.Data) != "msg" {
		t.Fatalf("Unexpected data: %q", msg.Data)
	}
}

func TestJetStreamPullSubscribeFetchContext(t *testing.T) {
	withJSCluster(t, "PULLCTX", 3, testJetStreamFetchContext)