Function TestGetLastMsgForSubject

jetstream/test/stream_test.go:1035–1157

(t *testing.T)

}

func TestGetLastMsgForSubject(t *testing.T) {
	tests := []struct {
		name         string
		subject      string
		expectedData string
		allowDirect  bool
		timeout      time.Duration
		withError    error
	}{
		{
			name:         "get existing msg",
			subject:      "*.A",
			expectedData: "msg 4 on subject A",
			timeout:      5 * time.Second,
		},
		{
			name:         "with empty context",
			subject:      "*.A",
			expectedData: "msg 4 on subject A",
		},
		{
			name:         "get last msg from stream",
			subject:      ">",
			expectedData: "msg 4 on subject B",
			timeout:      5 * time.Second,
		},
		{
			name:      "no messages on subject",
			subject:   "*.Z",
			withError: jetstream.ErrMsgNotFound,
		},
		{
			name:      "context timeout",
			subject:   "*.A",
			timeout:   1 * time.Microsecond,
			withError: context.DeadlineExceeded,
		},
	}

	srv := RunBasicJetStreamServer()
	defer shutdownJSServerAndRemoveStorage(t, srv)
	nc, err := nats.Connect(srv.ClientURL())
	if err != nil {
		t.Fatalf("Unexpected error: %v", err)
	}

	js, err := jetstream.New(nc)
	if err != nil {
		t.Fatalf("Unexpected error: %v", err)
	}
	defer nc.Close()

	s1, err := js.CreateStream(context.Background(), jetstream.StreamConfig{Name: "foo", Subjects: []string{"FOO.*"}, Description: "desc"})
	if err != nil {
		t.Fatalf("Unexpected error: %v", err)
	}
	for i := 1; i < 5; i++ {
		if _, err := js.Publish(context.Background(), "FOO.A", []byte(fmt.Sprintf("msg %d on subject A", i))); err != nil {