Repository: nats-io/nats.go
Pull request #991 (2022-06-07 13:45:02)

Align RePublish with server and add HeadersOnly, also add in AllowDirect option

Signed-off-by: Derek Collison <derek@nats.io>


Patch:
diff --git a/js.go b/js.go
index b703d3d..7785e49 100644
--- a/js.go
+++ b/js.go
@@ -230,8 +230,6 @@ type jsOpts struct {
 	shouldTrace bool
 	// purgeOpts contains optional stream purge options
 	purgeOpts *StreamPurgeRequest
-	// streamInfoOpts contains optional stream info options
-	streamInfoOpts *StreamInfoRequest
 }
 
 const (
@@ -303,11 +301,6 @@ func (s *StreamPurgeRequest) configureJSContext(js *jsOpts) error {
 	return nil
 }
 
-func (s *StreamInfoRequest) configureJSContext(js *jsOpts) error {
-	js.streamInfoOpts = s
-	return nil
-}
-
 // APIPrefix changes the default prefix used for the JetStream API.
 func APIPrefix(pre string) JSOpt {
 	return jsOptFn(func(js *jsOpts) error {
diff --git a/jsm.go b/jsm.go
index d08d0e3..8b45619 100644
--- a/jsm.go
+++ b/jsm.go
@@ -1,4 +1,4 @@
-// Copyright 2021 The NATS Authors
+// Copyright 2021-2022 The NATS Authors
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
 // You may obtain a copy of the License at
@@ -100,15 +100,18 @@ type StreamConfig struct {
 	DenyDelete        bool            `json:"deny_delete,omitempty"`
 	DenyPurge         bool            `json:"deny_purge,omitempty"`
 	AllowRollup       bool            `json:"allow_rollup_hdrs,omitempty"`
+	AllowDirect       bool            `json:"allow_direct,omitempty"`
 
 	// Allow republish of the message after being sequenced and stored.
-	RePublish *SubjectMapping `json:"republish,omitempty"`
+	RePublish *RePublish `json:"republish,omitempty"`
 }
 
-// SubjectMapping allows a source subject to be mapped to a destination subject for republishing.
-type SubjectMapping struct {
+// RePublish is for republishing messages once committed to a stream. The original
+// subject cis remapped from the subject pattern to the destination pattern.
+type RePublish struct {
 	Source      string `json:"src,omitempty"`
 	Destination string `json:"dest"`
+	HeadersOnly bool   `json:"headers_only,omitempty"`
 }
 
 // Placement is used to guide placement of streams in clustered JetStream.
@@ -634,13 +637,7 @@ func (js *js) AddStream(cfg *StreamConfig, opts ...JSOpt) (*StreamInfo, error) {
 	return resp.StreamInfo, nil
 }
 
-type (
-	// StreamInfoRequest contains additional option to return details about messages deleted from a stream
-	StreamInfoRequest struct {
-		DeletedDetails bool `json:"deleted_details,omitempty"`
-	}
-	streamInfoResponse = streamCreateResponse
-)
+type streamInfoResponse = streamCreateResponse
 
 func (js *js) StreamInfo(stream string, opts ...JSOpt) (*StreamInfo, error) {
 	if err := checkStreamName(stream); err != nil {
@@ -653,15 +650,9 @@ func (js *js) StreamInfo(stream string, opts ...JSOpt) (*StreamInfo, error) {
 	if cancel != nil {
 		defer cancel()
 	}
-	var req []byte
-	if o.streamInfoOpts != nil {
-		if req, err = json.Marshal(o.streamInfoOpts); err != nil {
-			return nil, err
-		}
-	}
-	siSubj := js.apiSubj(fmt.Sprintf(apiStreamInfoT, stream))
 
-	r, err := js.apiRequestWithContext(o.ctx, siSubj, req)
+	csSubj := js.apiSubj(fmt.Sprintf(apiStreamInfoT, stream))
+	r, err := js.apiRequestWithContext(o.ctx, csSubj, nil)
 	if err != nil {
 		return nil, err
 	}
@@ -698,15 +689,13 @@ type StreamSourceInfo struct {
 
 // StreamState is information about the given stream.
 type StreamState struct {
-	Msgs       uint64    `json:"messages"`
-	Bytes      uint64    `json:"bytes"`
-	FirstSeq   uint64    `json:"first_seq"`
-	FirstTime  time.Time `json:"first_ts"`
-	LastSeq    uint64    `json:"last_seq"`
-	LastTime   time.Time `json:"last_ts"`
-	Consumers  int       `json:"consumer_count"`
-	Deleted    []uint64  `json:"deleted"`
-	NumDeleted int       `json:"num_deleted"`
+	Msgs      uint64    `json:"messages"`
+	Bytes     uint64    `json:"bytes"`
+	FirstSeq  uint64    `json:"first_seq"`
+	FirstTime time.Time `json:"first_ts"`
+	LastSeq   uint64    `json:"last_seq"`
+	LastTime  time.Time `json:"last_ts"`
+	Consumers int       `json:"consumer_count"`
 }
 
 // ClusterInfo shows information about the underlying set of servers


Test patch:
diff --git a/test/js_test.go b/test/js_test.go
index 59730ac..0328e06 100644
--- a/test/js_test.go
+++ b/test/js_test.go
@@ -1702,110 +1702,6 @@ func TestPurgeStream(t *testing.T) {
 	}
 }
 
-func TestStreamInfo(t *testing.T) {
-	testData := []string{"one", "two", "three", "four"}
-
-	tests := []struct {
-		name                   string
-		stream                 string
-		req                    *nats.StreamInfoRequest
-		withError              error
-		expectedDeletedDetails []uint64
-	}{
-		{
-			name:   "empty request body",
-			stream: "foo",
-		},
-		{
-			name:   "with deleted details",
-			stream: "foo",
-			req: &nats.StreamInfoRequest{
-				DeletedDetails: true,
-			},
-			expectedDeletedDetails: []uint64{2, 4},
-		},
-		{
-			name:   "with deleted details set to false",
-			stream: "foo",
-			req: &nats.StreamInfoRequest{
-				DeletedDetails: false,
-			},
-		},
-		{
-			name:      "empty stream name",
-			stream:    "",
-			withError: nats.ErrStreamNameRequired,
-		},
-		{
-			name:      "invalid stream name",
-			stream:    "bad.stream.name",
-			withError: nats.ErrInvalidStreamName,
-		},
-		{
-			name:      "stream not found",
-			stream:    "bar",
-			withError: nats.ErrStreamNotFound,
-		},
-	}
-	for _, test := range tests {
-		t.Run(test.name, func(t *testing.T) {
-			s := RunBasicJetStreamServer()
-			defer shutdownJSServerAndRemoveStorage(t, s)
-
-			nc, js := jsClient(t, s)
-			defer nc.Close()
-
-			_, err := js.AddStream(&nats.StreamConfig{
-				Name:     "foo",
-				Subjects: []string{"foo.A"},
-			})
-			if err != nil {
-				t.Fatalf("Unexpected error: %v", err)
-			}
-
-			for _, msg := range testData {
-				if _, err := js.Publish("foo.A", []byte(msg)); err != nil {
-					t.Fatalf("Unexpected error during publish: %v", err)
-				}
-			}
-			if err := js.DeleteMsg("foo", 2); err != nil {
-				t.Fatalf("Unexpected error while deleting message from stream: %v", err)
-			}
-			if err := js.DeleteMsg("foo", 4); err != nil {
-				t.Fatalf("Unexpected error while deleting message from stream: %v", err)
-			}
-
-			var streamInfo *nats.StreamInfo
-			if test.req != nil {
-				streamInfo, err = js.StreamInfo(test.stream, test.req)
-			} else {
-				streamInfo, err = js.StreamInfo(test.stream)
-			}
-			if test.withError != nil {
-				if err == nil {
-					t.Fatal("Expected error, got nil")
-				}
-				if !errors.Is(err, test.withError) {
-					t.Fatalf("Expected error: '%s'; got '%s'", test.withError, err)
-				}
-				return
-			}
-			if err != nil {
-				t.Fatalf("Unexpected error: %v", err)
-			}
-			if streamInfo.Config.Name != "foo" {
-				t.Fatalf("Invalid stream name in StreamInfo response: want: 'foo'; got: '%s'", streamInfo.Config.Name)
-			}
-			if streamInfo.State.NumDeleted != 2 {
-				t.Fatalf("Invalid value for num_deleted in state: want: 2; got: %d", streamInfo.State.NumDeleted)
-			}
-			if !reflect.DeepEqual(test.expectedDeletedDetails, streamInfo.State.Deleted) {
-				t.Fatalf("Invalid value for deleted msgs in state: want: %v; got: %v", test.expectedDeletedDetails, streamInfo.State.Deleted)
-			}
-		})
-	}
-}
-
 func TestJetStreamManagement_GetMsg(t *testing.T) {
 	t.Run("1-node", func(t *testing.T) {
 		withJSServer(t, testJetStreamManagement_GetMsg)
@@ -2465,7 +2361,7 @@ func TestJetStreamCrossAccountMirrorsAndSources(t *testing.T) {
 		Name:    sourceName,
 		Storage: nats.FileStorage,
 		Sources: []*nats.StreamSource{
-			{
+			&nats.StreamSource{
 				Name: publishSubj,
 				External: &nats.ExternalStream{
 					APIPrefix:     "RI.JS.API",
@@ -7060,7 +6956,7 @@ func TestJetStreamRePublish(t *testing.T) {
 		Name:     "RP",
 		Storage:  nats.MemoryStorage,
 		Subjects: []string{"foo", "bar", "baz"},
-		RePublish: &nats.SubjectMapping{
+		RePublish: &nats.RePublish{
 			Source:      ">",
 			Destination: "RP.>",
 		},