Repository: nats-io/nats-server
Pull request #2313 (2021-06-24 20:22:58)

Added in Stream get last message by subject.
This is to aid in K/V overlay for simple Get ops vs creating a full blown consumer.

Signed-off-by: Derek Collison <derek@nats.io>

/cc @nats-io/core



Patch:
diff --git a/server/filestore.go b/server/filestore.go
index 9dbfd54b..deb17280 100644
--- a/server/filestore.go
+++ b/server/filestore.go
@@ -3143,6 +3143,21 @@ func (fs *fileStore) LoadMsg(seq uint64) (string, []byte, []byte, int64, error)
 	return _EMPTY_, nil, nil, 0, err
 }
 
+// LoadLastMsg will return the last message we have that matches a given subject.
+// The subject can be a wildcard.
+func (fs *fileStore) LoadLastMsg(subject string) (subj string, seq uint64, hdr, msg []byte, ts int64, err error) {
+	var sm *fileStoredMsg
+	if subject == _EMPTY_ || subject == fwcs {
+		sm, _ = fs.msgForSeq(fs.lastSeq())
+	} else if ss := fs.FilteredState(1, subject); ss.Msgs > 0 {
+		sm, _ = fs.msgForSeq(ss.Last)
+	}
+	if sm == nil {
+		return _EMPTY_, 0, nil, nil, 0, ErrStoreMsgNotFound
+	}
+	return sm.subj, sm.seq, sm.hdr, sm.msg, sm.ts, nil
+}
+
 // Type returns the type of the underlying store.
 func (fs *fileStore) Type() StorageType {
 	return FileStorage
diff --git a/server/jetstream_api.go b/server/jetstream_api.go
index 94a19293..3a619059 100644
--- a/server/jetstream_api.go
+++ b/server/jetstream_api.go
@@ -494,7 +494,8 @@ const JSApiMetaServerRemoveResponseType = "io.nats.jetstream.api.v1.meta_server_
 
 // JSApiMsgGetRequest get a message request.
 type JSApiMsgGetRequest struct {
-	Seq uint64 `json:"seq"`
+	Seq     uint64 `json:"seq,omitempty"`
+	LastFor string `json:"last_by_subj,omitempty"`
 }
 
 type JSApiMsgGetResponse struct {
@@ -2418,6 +2419,13 @@ func (s *Server) jsMsgGetRequest(sub *subscription, c *client, subject, reply st
 		return
 	}
 
+	// Check that we do not have both options set.
+	if req.Seq > 0 && req.LastFor != _EMPTY_ || req.Seq == 0 && req.LastFor == _EMPTY_ {
+		resp.Error = ApiErrors[JSBadRequestErr]
+		s.sendAPIErrResponse(ci, acc, subject, reply, string(msg), s.jsonResponse(&resp))
+		return
+	}
+
 	mset, err := acc.lookupStream(stream)
 	if err != nil {
 		resp.Error = ApiErrors[JSStreamNotFoundErr]
@@ -2425,7 +2433,16 @@ func (s *Server) jsMsgGetRequest(sub *subscription, c *client, subject, reply st
 		return
 	}
 
-	subj, hdr, msg, ts, err := mset.store.LoadMsg(req.Seq)
+	var subj string
+	var hdr []byte
+	var ts int64
+	seq := req.Seq
+
+	if req.Seq > 0 {
+		subj, hdr, msg, ts, err = mset.store.LoadMsg(req.Seq)
+	} else {
+		subj, seq, hdr, msg, ts, err = mset.store.LoadLastMsg(req.LastFor)
+	}
 	if err != nil {
 		resp.Error = ApiErrors[JSNoMessageFoundErr]
 		s.sendAPIErrResponse(ci, acc, subject, reply, string(msg), s.jsonResponse(&resp))
@@ -2433,7 +2450,7 @@ func (s *Server) jsMsgGetRequest(sub *subscription, c *client, subject, reply st
 	}
 	resp.Message = &StoredMsg{
 		Subject:  subj,
-		Sequence: req.Seq,
+		Sequence: seq,
 		Header:   hdr,
 		Data:     msg,
 		Time:     time.Unix(0, ts).UTC(),
diff --git a/server/memstore.go b/server/memstore.go
index 632960b4..f277431d 100644
--- a/server/memstore.go
+++ b/server/memstore.go
@@ -255,7 +255,10 @@ func (ms *memStore) GetSeqFromTime(t time.Time) uint64 {
 func (ms *memStore) FilteredState(sseq uint64, subj string) SimpleState {
 	ms.mu.RLock()
 	defer ms.mu.RUnlock()
+	return ms.filteredStateLocked(sseq, subj)
+}
 
+func (ms *memStore) filteredStateLocked(sseq uint64, subj string) SimpleState {
 	var ss SimpleState
 
 	if sseq < ms.state.FirstSeq {
@@ -569,11 +572,31 @@ func (ms *memStore) LoadMsg(seq uint64) (string, []byte, []byte, int64, error) {
 		if seq <= last {
 			err = ErrStoreMsgNotFound
 		}
-		return "", nil, nil, 0, err
+		return _EMPTY_, nil, nil, 0, err
 	}
 	return sm.subj, sm.hdr, sm.msg, sm.ts, nil
 }
 
+// LoadLastMsg will return the last message we have that matches a given subject.
+// The subject can be a wildcard.
+func (ms *memStore) LoadLastMsg(subject string) (subj string, seq uint64, hdr, msg []byte, ts int64, err error) {
+	var sm *storedMsg
+	var ok bool
+
+	ms.mu.RLock()
+	defer ms.mu.RUnlock()
+
+	if subject == _EMPTY_ || subject == fwcs {
+		sm, ok = ms.msgs[ms.state.LastSeq]
+	} else if ss := ms.filteredStateLocked(1, subject); ss.Msgs > 0 {
+		sm, ok = ms.msgs[ss.Last]
+	}
+	if !ok || sm == nil {
+		return _EMPTY_, 0, nil, nil, 0, ErrStoreMsgNotFound
+	}
+	return sm.subj, sm.seq, sm.hdr, sm.msg, sm.ts, nil
+}
+
 // RemoveMsg will remove the message from this store.
 // Will return the number of bytes removed.
 func (ms *memStore) RemoveMsg(seq uint64) (bool, error) {
diff --git a/server/store.go b/server/store.go
index f39ff7c3..343a7948 100644
--- a/server/store.go
+++ b/server/store.go
@@ -71,6 +71,7 @@ type StreamStore interface {
 	StoreRawMsg(subject string, hdr, msg []byte, seq uint64, ts int64) error
 	SkipMsg() uint64
 	LoadMsg(seq uint64) (subject string, hdr, msg []byte, ts int64, err error)
+	LoadLastMsg(subject string) (subj string, seq uint64, hdr, msg []byte, ts int64, err error)
 	RemoveMsg(seq uint64) (bool, error)
 	EraseMsg(seq uint64) (bool, error)
 	Purge() (uint64, error)


Test patch:
diff --git a/server/jetstream_test.go b/server/jetstream_test.go
index 99dab095..d140668c 100644
--- a/server/jetstream_test.go
+++ b/server/jetstream_test.go
@@ -10863,6 +10863,106 @@ func TestJetStreamMaxMsgsPerSubject(t *testing.T) {
 	}
 }
 
+func TestJetStreamGetLastMsgBySubject(t *testing.T) {
+	for _, st := range []StorageType{FileStorage, MemoryStorage} {
+		t.Run(st.String(), func(t *testing.T) {
+			c := createJetStreamClusterExplicit(t, "JSC", 3)
+			defer c.shutdown()
+
+			nc, js := jsClientConnect(t, c.randomServer())
+			defer nc.Close()
+
+			cfg := StreamConfig{
+				Name:       "KV",
+				Subjects:   []string{"kv.>"},
+				Storage:    st,
+				Replicas:   2,
+				MaxMsgsPer: 20,
+			}
+
+			req, err := json.Marshal(cfg)
+			if err != nil {
+				t.Fatalf("Unexpected error: %v", err)
+			}
+			// Do manually for now.
+			nc.Request(fmt.Sprintf(JSApiStreamCreateT, cfg.Name), req, time.Second)
+			si, err := js.StreamInfo("KV")
+			if err != nil {
+				t.Fatalf("Unexpected error: %v", err)
+			}
+			if si == nil || si.Config.Name != "KV" {
+				t.Fatalf("StreamInfo is not correct %+v", si)
+			}
+
+			for i := 0; i < 1000; i++ {
+				msg := []byte(fmt.Sprintf("VAL-%d", i+1))
+				js.PublishAsync("kv.foo", msg)
+				js.PublishAsync("kv.bar", msg)
+				js.PublishAsync("kv.baz", msg)
+			}
+			select {
+			case <-js.PublishAsyncComplete():
+			case <-time.After(5 * time.Second):
+				t.Fatalf("Did not receive completion signal")
+			}
+
+			// Check that if both set that errors.
+			b, _ := json.Marshal(JSApiMsgGetRequest{LastFor: "kv.foo", Seq: 950})
+			rmsg, err := nc.Request(fmt.Sprintf(JSApiMsgGetT, "KV"), b, time.Second)
+			if err != nil {
+				t.Fatalf("Unexpected error: %v", err)
+			}
+			var resp JSApiMsgGetResponse
+			err = json.Unmarshal(rmsg.Data, &resp)
+			if err != nil {
+				t.Fatalf("Could not parse stream message: %v", err)
+			}
+			if resp.Error == nil {
+				t.Fatalf("Expected an error when both are set, got %+v", resp.Error)
+			}
+
+			// Need to do stream GetMsg by hand for now until Go client support lands.
+			getLast := func(subject string) *StoredMsg {
+				t.Helper()
+				req := &JSApiMsgGetRequest{LastFor: subject}
+				b, _ := json.Marshal(req)
+				rmsg, err := nc.Request(fmt.Sprintf(JSApiMsgGetT, "KV"), b, time.Second)
+				if err != nil {
+					t.Fatalf("Unexpected error: %v", err)
+				}
+				var resp JSApiMsgGetResponse
+				err = json.Unmarshal(rmsg.Data, &resp)
+				if err != nil {
+					t.Fatalf("Could not parse stream message: %v", err)
+				}
+				if resp.Message == nil || resp.Error != nil {
+					t.Fatalf("Did not receive correct response: %+v", resp.Error)
+				}
+				return resp.Message
+			}
+			// Do basic checks.
+			basicCheck := func(subject string, expectedSeq uint64) {
+				sm := getLast(subject)
+				if sm == nil {
+					t.Fatalf("Expected a message but got none")
+				} else if string(sm.Data) != "VAL-1000" {
+					t.Fatalf("Wrong message payload, wanted %q but got %q", "VAL-1000", sm.Data)
+				} else if sm.Sequence != expectedSeq {
+					t.Fatalf("Wrong message sequence, wanted %d but got %d", expectedSeq, sm.Sequence)
+				} else if !subjectIsSubsetMatch(sm.Subject, subject) {
+					t.Fatalf("Wrong subject, wanted %q but got %q", subject, sm.Subject)
+				}
+			}
+
+			basicCheck("kv.foo", 2998)
+			basicCheck("kv.bar", 2999)
+			basicCheck("kv.baz", 3000)
+			basicCheck("kv.*", 3000)
+			basicCheck(">", 3000)
+		})
+	}
+}
+
 func TestJetStreamFilteredConsumersWithWiderFilter(t *testing.T) {
 	s := RunBasicJetStreamServer()
 	defer s.Shutdown()