GODRIVER-1383 ChangeStream and Cursor documentation (#229)

Author: Divjot Arora

mongo/change_stream.go

@@ -31,19 +31,18 @@ const errorInterrupted int32 = 11601
 const errorCappedPositionLost int32 = 136
 const errorCursorKilled int32 = 237
 
-// ErrMissingResumeToken indicates that a change stream notification from the server did not
-// contain a resume token.
+// ErrMissingResumeToken indicates that a change stream notification from the server did not contain a resume token.
 var ErrMissingResumeToken = errors.New("cannot provide resume functionality when the resume token is missing")
 
-// ErrNilCursor indicates that the cursor for the change stream is nil.
+// ErrNilCursor indicates that the underlying cursor for the change stream is nil.
 var ErrNilCursor = errors.New("cursor is nil")
 
-// ChangeStream instances iterate a stream of change documents. Each document can be decoded via the
-// Decode method. Resume tokens should be retrieved via the ResumeToken method and can be stored to
-// resume the change stream at a specific point in time.
-//
-// A typical usage of the ChangeStream type would be:
+// ChangeStream is used to iterate over a stream of events. Each event can be decoded into a Go type via the Decode
+// method or accessed as raw BSON via the Current field. For more information about change streams, see
+// https://docs.mongodb.com/manual/changeStreams/.
 type ChangeStream struct {
+	// Current is the BSON bytes of the current event. This property is only valid until the next call to Next or
+	// TryNext. If continued access is required, a copy must be made.
 	Current bson.Raw
 
 	aggregate     *operation.Aggregate
@@ -405,15 +404,16 @@ func (cs *ChangeStream) replaceOptions(ctx context.Context, wireVersion *descrip
 	cs.options.SetStartAtOperationTime(nil)
 }
 
-// ID returns the cursor ID for this change stream.
+// ID returns the ID for this change stream, or 0 if the cursor has been closed or exhausted.
 func (cs *ChangeStream) ID() int64 {
 	if cs.cursor == nil {
 		return 0
 	}
 	return cs.cursor.ID()
 }
 
-// Decode will decode the current document into val.
+// Decode will unmarshal the current event document into val and return any errors from the unmarshalling process
+// without any modification. If val is nil or is a typed nil, an error will be returned.
 func (cs *ChangeStream) Decode(val interface{}) error {
 	if cs.cursor == nil {
 		return ErrNilCursor
@@ -422,7 +422,7 @@ func (cs *ChangeStream) Decode(val interface{}) error {
 	return bson.UnmarshalWithRegistry(cs.registry, cs.Current, val)
 }
 
-// Err returns the current error.
+// Err returns the last error seen by the change stream, or nil if no errors has occurred.
 func (cs *ChangeStream) Err() error {
 	if cs.err != nil {
 		return replaceErrors(cs.err)
@@ -434,7 +434,8 @@ func (cs *ChangeStream) Err() error {
 	return replaceErrors(cs.cursor.Err())
 }
 
-// Close closes this cursor.
+// Close closes this change stream and the underlying cursor. Next and TryNext must not be called after Close has been
+// called. Close is idempotent. After the first call, any subsequent calls will not change the state.
 func (cs *ChangeStream) Close(ctx context.Context) error {
 	if ctx == nil {
 		ctx = context.Background()
@@ -451,26 +452,34 @@ func (cs *ChangeStream) Close(ctx context.Context) error {
 	return cs.Err()
 }
 
-// ResumeToken returns the last cached resume token for this change stream.
+// ResumeToken returns the last cached resume token for this change stream, or nil if a resume token has not been
+// stored.
 func (cs *ChangeStream) ResumeToken() bson.Raw {
 	return cs.resumeToken
 }
 
-// Next gets the next result from this change stream. Returns true if there were no errors and the next
-// result is available for decoding. Next blocks until an event is available for decoding or ctx expires.
-// If the given context expires during execution, the stream's error will be set and the change stream may be in an
-// invalid state and should be re-created. If Next returns false, it must not be called again.
+// Next gets the next event for this change stream. It returns true if there were no errors and the next event document
+// is available.
+//
+// Next blocks until an event is available, an error occurs, or ctx expires. If ctx expires, the error
+// will be set to ctx.Err(). In an error case, Next will return false.
+//
+// If Next returns false, subsequent calls will also return false.
 func (cs *ChangeStream) Next(ctx context.Context) bool {
 	return cs.next(ctx, false)
 }
 
-// TryNext attempts to get the next result from this change stream. It returns true if there were no errors and the next
-// result is available for decoding. It returns false if the change stream was closed by the server, there was an
-// error getting more results from the server, the server returned an empty batch of events, or the given context expires.
-// If an error occurred or the stream was closed (can be checked with cs.Err() != nil || cs.ID() == 0), TryNext must
-// not be called again. If the given context expires during execution, the stream's error will be set and the change
-// stream may be in an invalid state and should be re-created.
-// Added in version 1.2.0.
+// TryNext attempts to get the next event for this change stream. It returns true if there were no errors and the next
+// event document is available.
+//
+// TryNext returns false if the change stream is closed by the server, an error occurs when getting changes from the
+// server, the next change is not yet available, or ctx expires. If ctx expires, the error will be set to ctx.Err().
+//
+// If TryNext returns false and an error occurred or the change stream was closed
+// (i.e. cs.Err() != nil || cs.ID() == 0), subsequent attempts will also return false. Otherwise, it is safe to call
+// TryNext again until a change is available.
+//
+// This method requires driver version >= 1.2.0.
 func (cs *ChangeStream) TryNext(ctx context.Context) bool {
 	return cs.next(ctx, true)
 }
@@ -549,11 +558,11 @@ func (cs *ChangeStream) emptyBatch() bool {
 	return cs.cursor.Batch().Empty()
 }
 
-// StreamType represents the type of a change stream.
+// StreamType represents the cluster type against which a ChangeStream was created.
 type StreamType uint8
 
 // These constants represent valid change stream types. A change stream can be initialized over a collection, all
-// collections in a database, or over a whole client.
+// collections in a database, or over a cluster.
 const (
 	CollectionStream StreamType = iota
 	DatabaseStream

mongo/crud_examples_test.go

@@ -10,6 +10,7 @@ import (
 	"context"
 	"fmt"
 	"log"
+	"sync"
 	"time"
 
 	"go.mongodb.org/mongo-driver/bson"
@@ -556,3 +557,150 @@ func ExampleClient_StartSession_withTransaction() {
 	}
 	fmt.Printf("result: %v\n", result)
 }
+
+// Cursor examples
+
+func ExampleCursor_All() {
+	var cursor *mongo.Cursor
+
+	var results []bson.M
+	if err := cursor.All(context.TODO(), &results); err != nil {
+		log.Fatal(err)
+	}
+	fmt.Println(results)
+}
+
+func ExampleCursor_Next() {
+	var cursor *mongo.Cursor
+	defer cursor.Close(context.TODO())
+
+	// Iterate the cursor and print out each document until the cursor is exhausted or there is an error getting the
+	// next document.
+	for cursor.Next(context.TODO()) {
+		// A new result variable should be declared for each document.
+		var result bson.M
+		if err := cursor.Decode(&result); err != nil {
+			log.Fatal(err)
+		}
+		fmt.Println(result)
+	}
+	if err := cursor.Err(); err != nil {
+		log.Fatal(err)
+	}
+}
+
+func ExampleCursor_TryNext() {
+	var cursor *mongo.Cursor
+	defer cursor.Close(context.TODO())
+
+	// Iterate the cursor and print out each document until the cursor is exhausted or there is an error getting the
+	// next document.
+	for {
+		if cursor.TryNext(context.TODO()) {
+			// A new result variable should be declared for each document.
+			var result bson.M
+			if err := cursor.Decode(&result); err != nil {
+				log.Fatal(err)
+			}
+			fmt.Println(result)
+			continue
+		}
+
+		// If TryNext returns false, the next document is not yet available, the cursor was exhausted and was closed, or
+		// an error occured. TryNext should only be called again for the empty batch case.
+		if err := cursor.Err(); err != nil {
+			log.Fatal(err)
+		}
+		if cursor.ID() == 0 {
+			break
+		}
+	}
+}
+
+// ChangeStream examples
+
+func ExampleChangeStream_Next() {
+	var stream *mongo.ChangeStream
+	defer stream.Close(context.TODO())
+
+	// Iterate the change stream and print out each event.
+	// Because the Next call blocks until an event is available, another way to iterate the change stream is to call
+	// Next in a goroutine and pass in a context that can be cancelled to abort the call.
+
+	for stream.Next(context.TODO()) {
+		// A new event variable should be declared for each event.
+		var event bson.M
+		if err := stream.Decode(&event); err != nil {
+			log.Fatal(err)
+		}
+		fmt.Println(event)
+	}
+	if err := stream.Err(); err != nil {
+		log.Fatal(err)
+	}
+}
+
+func ExampleChangeStream_TryNext() {
+	var stream *mongo.ChangeStream
+	defer stream.Close(context.TODO())
+
+	// Iterate the change stream and print out each event until the change stream is closed by the server or there is an
+	// error getting the next event.
+	for {
+		if stream.TryNext(context.TODO()) {
+			// A new event variable should be declared for each event.
+			var event bson.M
+			if err := stream.Decode(&event); err != nil {
+				log.Fatal(err)
+			}
+			fmt.Println(event)
+			continue
+		}
+
+		// If TryNext returns false, the next change is not yet available, the change stream was closed by the server,
+		// or an error occurred. TryNext should only be called again for the empty batch case.
+		if err := stream.Err(); err != nil {
+			log.Fatal(err)
+		}
+		if stream.ID() == 0 {
+			break
+		}
+	}
+}
+
+func ExampleChangeStream_ResumeToken() {
+	var client *mongo.Client
+	var stream *mongo.ChangeStream // assume stream was created via client.Watch()
+
+	cancelCtx, cancel := context.WithCancel(context.TODO())
+	defer cancel()
+	var wg sync.WaitGroup
+	wg.Add(1)
+
+	// Run a goroutine to process events.
+	go func() {
+		for stream.Next(cancelCtx) {
+			fmt.Println(stream.Current)
+		}
+		wg.Done()
+	}()
+
+	// Assume client needs to be disconnected. Cancel the context being used by the goroutine to abort any
+	// in-progres Next calls and wait for the goroutine to exit.
+	cancel()
+	wg.Wait()
+
+	// Before disconnecting the client, store the last seen resume token for the change stream.
+	resumeToken := stream.ResumeToken()
+	_ = client.Disconnect(context.TODO())
+
+	// Once a new client is created, the change stream can be re-created. Specify resumeToken as the ResumeAfter option
+	// so only events that occurred after resumeToken will be returned.
+	var newClient *mongo.Client
+	opts := options.ChangeStream().SetResumeAfter(resumeToken)
+	newStream, err := newClient.Watch(context.TODO(), mongo.Pipeline{}, opts)
+	if err != nil {
+		log.Fatal(err)
+	}
+	defer newStream.Close(context.TODO())
+}