Repository: nats-io/nats-server
Pull request #2596 (2021-10-06 01:55:25)

Allow consumers to request only headers to be delivered
Signed-off-by: Derek Collison <derek@nats.io>

/cc @nats-io/core



Patch:
diff --git a/server/consumer.go b/server/consumer.go
index f634ccb1..a167c067 100644
--- a/server/consumer.go
+++ b/server/consumer.go
@@ -66,6 +66,7 @@ type ConsumerConfig struct {
 	MaxAckPending   int           `json:"max_ack_pending,omitempty"`
 	Heartbeat       time.Duration `json:"idle_heartbeat,omitempty"`
 	FlowControl     bool          `json:"flow_control,omitempty"`
+	HeadersOnly     bool          `json:"headers_only,omitempty"`
 
 	// Don't add to general clients.
 	Direct bool `json:"direct,omitempty"`
@@ -2331,6 +2332,25 @@ func (o *consumer) deliverMsg(dsubj, subj string, hdr, msg []byte, seq, dc uint6
 	dseq := o.dseq
 	o.dseq++
 
+	// If headers only do not send msg payload.
+	// Add in msg size itself as header.
+	if o.cfg.HeadersOnly {
+		bb := bytes.NewBuffer(hdr)
+		if hdr == nil {
+			bb.WriteString(hdrLine)
+		} else {
+			bb.Truncate(len(hdr) - LEN_CR_LF)
+		}
+		bb.WriteString(JSMsgSize)
+		bb.WriteString(": ")
+		bb.WriteString(strconv.FormatInt(int64(len(msg)), 10))
+		bb.WriteString(CR_LF)
+		bb.WriteString(CR_LF)
+		hdr = bb.Bytes()
+		// Cancel msg payload
+		msg = nil
+	}
+
 	pmsg := &jsPubMsg{dsubj, subj, o.ackReply(seq, dseq, dc, ts, o.adjustedPending()), hdr, msg, o, seq, nil}
 	if o.maxpb > 0 {
 		o.pbytes += pmsg.size()
diff --git a/server/stream.go b/server/stream.go
index 0712cefb..23df6f88 100644
--- a/server/stream.go
+++ b/server/stream.go
@@ -222,6 +222,7 @@ const (
 	JSLastStreamSeq       = "Nats-Last-Stream"
 	JSConsumerStalled     = "Nats-Consumer-Stalled"
 	JSMsgRollup           = "Nats-Rollup"
+	JSMsgSize             = "Nats-Msg-Size"
 )
 
 // Rollups, can be subject only or all messages.


Test patch:
diff --git a/server/jetstream_test.go b/server/jetstream_test.go
index 4a9c1771..958d96c0 100644
--- a/server/jetstream_test.go
+++ b/server/jetstream_test.go
@@ -13330,6 +13330,52 @@ func TestJetStreamDisabledLimitsEnforcement(t *testing.T) {
 	require_Error(t, err)
 }
 
+func TestJetStreamConsumerNoMsgPayload(t *testing.T) {
+	s := RunBasicJetStreamServer()
+	defer s.Shutdown()
+
+	config := s.JetStreamConfig()
+	if config != nil {
+		defer removeDir(t, config.StoreDir)
+	}
+
+	nc, js := jsClientConnect(t, s)
+	defer nc.Close()
+
+	_, err := js.AddStream(&nats.StreamConfig{Name: "S"})
+	require_NoError(t, err)
+
+	msg := nats.NewMsg("S")
+	msg.Header.Set("name", "derek")
+	msg.Data = bytes.Repeat([]byte("A"), 128)
+	for i := 0; i < 10; i++ {
+		msg.Reply = _EMPTY_ // Fixed in Go client but not in embdedded on yet.
+		_, err = js.PublishMsgAsync(msg)
+		require_NoError(t, err)
+	}
+
+	mset, err := s.GlobalAccount().lookupStream("S")
+	require_NoError(t, err)
+
+	// Now create our consumer with no payload option.
+	_, err = mset.addConsumer(&ConsumerConfig{DeliverSubject: "_d_", Durable: "d22", HeadersOnly: true})
+	require_NoError(t, err)
+
+	sub, err := js.SubscribeSync("S", nats.Durable("d22"))
+	require_NoError(t, err)
+
+	for i := 0; i < 10; i++ {
+		m, err := sub.NextMsg(time.Second)
+		require_NoError(t, err)
+		if len(m.Data) > 0 {
+			t.Fatalf("Expected no payload")
+		}
+		if ms := m.Header.Get(JSMsgSize); ms != "128" {
+			t.Fatalf("Expected a header with msg size, got %q", ms)
+		}
+	}
+}
+
 ///////////////////////////////////////////////////////////////////////////
 // Simple JetStream Benchmarks
 ///////////////////////////////////////////////////////////////////////////