Function TestJetStreamDirectGetMsg

test/js_test.go:9873–9985

(t *testing.T)

}

func TestJetStreamDirectGetMsg(t *testing.T) {
	// Using standalone server here, we are testing the client side API, not
	// the server feature, which has tests checking it works in cluster mode.
	s := RunBasicJetStreamServer()
	defer shutdownJSServerAndRemoveStorage(t, s)

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

	si, err := js.AddStream(&nats.StreamConfig{
		Name:     "DGM",
		Storage:  nats.MemoryStorage,
		Subjects: []string{"foo", "bar"},
	})
	if err != nil {
		t.Fatalf("Error adding stream: %v", err)
	}

	send := func(subj, body string) {
		t.Helper()
		if _, err := js.Publish(subj, []byte(body)); err != nil {
			t.Fatalf("Error on publish: %v", err)
		}
	}

	send("foo", "a")
	send("foo", "b")
	send("foo", "c")
	send("bar", "d")
	send("foo", "e")

	// Without AllowDirect, we should get no responders
	if _, err := js.GetMsg("DGM", 1, nats.DirectGet()); err != nats.ErrNoResponders {
		t.Fatalf("Unexpected error: %v", err)
	}

	// Update stream:
	si.Config.AllowDirect = true
	si, err = js.UpdateStream(&si.Config)
	if err != nil {
		t.Fatalf("Error updating stream: %v", err)
	}
	if !si.Config.AllowDirect {
		t.Fatalf("AllowDirect should be true: %+v", si)
	}

	check := func(seq uint64, opt nats.JSOpt, useGetLast bool, expectedSubj string, expectedSeq uint64, expectedBody string) {
		t.Helper()

		var msg *nats.RawStreamMsg
		var err error
		if useGetLast {
			msg, err = js.GetLastMsg("DGM", expectedSubj, []nats.JSOpt{opt}...)
		} else {
			msg, err = js.GetMsg("DGM", seq, []nats.JSOpt{opt}...)
		}
		if err != nil {
			t.Fatalf("Unable to get message: %v", err)