Repository: nats-io/nats.go
Pull request #995 (2022-06-07 18:12:31)

Update object Put to avoid loosing last chunk
The old code might miss the last object if `Reader` return both a value and `EOF`

Extract from `Reader` doc :
> When Read encounters an error or end-of-file condition after successfully reading n > 0 bytes, it returns the number of bytes read. It may return the (non-nil) error from the same call or return the error (and n == 0) from a subsequent call. An instance of this general case is that a Reader returning a non-zero number of bytes at the end of the input stream may return either err == EOF or err == nil. The next Read should return 0, EOF.
>
> Callers should always process the n > 0 bytes returned before considering the error err. Doing so correctly handles I/O errors that happen after reading some bytes and also both of the allowed EOF behaviors.

derekcollison: Can we squash this done to one CL?
tinou98: You should be able to Squash & Merge from GitHub UI, otherwise I can Squash & Force Push on my branch
derekcollison: Would you mind squashing and force push?

Patch:
diff --git a/js.go b/js.go
index e34b297..8b1d663 100644
--- a/js.go
+++ b/js.go
@@ -97,8 +97,7 @@ type JetStream interface {
 	QueueSubscribeSync(subj, queue string, opts ...SubOpt) (*Subscription, error)
 
 	// PullSubscribe creates a Subscription that can fetch messages.
-	// See important note in Subscribe(). Additionally, for an ephemeral pull consumer, the "durable" value must be
-	// set to an empty string.
+	// See important note in Subscribe()
 	PullSubscribe(subj, durable string, opts ...SubOpt) (*Subscription, error)
 }
 
diff --git a/jsm.go b/jsm.go
index c9bcb2e..d08d0e3 100644
--- a/jsm.go
+++ b/jsm.go
@@ -1,4 +1,4 @@
-// Copyright 2021-2022 The NATS Authors
+// Copyright 2021 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,18 +100,15 @@ 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 *RePublish `json:"republish,omitempty"`
+	RePublish *SubjectMapping `json:"republish,omitempty"`
 }
 
-// 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 {
+// SubjectMapping allows a source subject to be mapped to a destination subject for republishing.
+type SubjectMapping 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.
diff --git a/object.go b/object.go
index f2646d8..d2124bb 100644
--- a/object.go
+++ b/object.go
@@ -372,37 +372,43 @@ func (obs *obs) Put(meta *ObjectMeta, r io.Reader, opts ...ObjectOpt) (*ObjectIn
 
 		// Actual read.
 		// TODO(dlc) - Deadline?
-		n, err := r.Read(chunk)
+		n, readErr := r.Read(chunk)
+
+		// Handle all non EOF errors
+		if readErr != nil && readErr != io.EOF {
+			purgePartial()
+			return nil, readErr
+		}
+
+		// Add chunk only if we received data
+		if n > 0 {
+			// Chunk processing.
+			m.Data = chunk[:n]
+			h.Write(m.Data)
+
+			// Send msg itself.
+			if _, err := js.PublishMsgAsync(m); err != nil {
+				purgePartial()
+				return nil, err
+			}
+			if err := getErr(); err != nil {
+				purgePartial()
+				return nil, err
+			}
+			// Update totals.
+			sent++
+			total += uint64(n)
+		}
 
 		// EOF Processing.
-		if err == io.EOF {
+		if readErr == io.EOF {
 			// Finalize sha.
 			sha := h.Sum(nil)
 			// Place meta info.
 			info.Size, info.Chunks = uint64(total), uint32(sent)
 			info.Digest = fmt.Sprintf(objDigestTmpl, base64.URLEncoding.EncodeToString(sha[:]))
 			break
-		} else if err != nil {
-			purgePartial()
-			return nil, err
-		}
-
-		// Chunk processing.
-		m.Data = chunk[:n]
-		h.Write(m.Data)
-
-		// Send msg itself.
-		if _, err := js.PublishMsgAsync(m); err != nil {
-			purgePartial()
-			return nil, err
-		}
-		if err := getErr(); err != nil {
-			purgePartial()
-			return nil, err
 		}
-		// Update totals.
-		sent++
-		total += uint64(n)
 	}
 
 	// Publish the metadata.


Test patch:
diff --git a/test/js_test.go b/test/js_test.go
index fd002dd..59730ac 100644
--- a/test/js_test.go
+++ b/test/js_test.go
@@ -7060,7 +7060,7 @@ func TestJetStreamRePublish(t *testing.T) {
 		Name:     "RP",
 		Storage:  nats.MemoryStorage,
 		Subjects: []string{"foo", "bar", "baz"},
-		RePublish: &nats.RePublish{
+		RePublish: &nats.SubjectMapping{
 			Source:      ">",
 			Destination: "RP.>",
 		},