Add mongo/changestreamopt

GODRIVER-272

Change-Id: I8dffbf535f7756dd18980bcc3bf4f4e2d28092fc

Author: Divjot Arora

examples/documentation_examples/examples.go

@@ -1166,7 +1166,7 @@ func ProjectionExamples(t *testing.T, db *mongo.Database) {
 	{
 		// Start Example 44
 
-		projection, err := mongo.Opt.Projection(bson.NewDocument(
+		projection := findopt.Projection(bson.NewDocument(
 			bson.EC.Int32("item", 1),
 			bson.EC.Int32("status", 1),
 		))
@@ -1177,7 +1177,7 @@ func ProjectionExamples(t *testing.T, db *mongo.Database) {
 			bson.NewDocument(
 				bson.EC.String("status", "A"),
 			),
-			findopt.Projection(projection),
+			projection,
 		)
 
 		// End Example 44
@@ -1206,7 +1206,7 @@ func ProjectionExamples(t *testing.T, db *mongo.Database) {
 	{
 		// Start Example 45
 
-		projection, err := mongo.Opt.Projection(bson.NewDocument(
+		projection := findopt.Projection(bson.NewDocument(
 			bson.EC.Int32("item", 1),
 			bson.EC.Int32("status", 1),
 			bson.EC.Int32("_id", 0),
@@ -1218,7 +1218,7 @@ func ProjectionExamples(t *testing.T, db *mongo.Database) {
 			bson.NewDocument(
 				bson.EC.String("status", "A"),
 			),
-			findopt.Projection(projection),
+			projection,
 		)
 
 		// End Example 45
@@ -1247,7 +1247,7 @@ func ProjectionExamples(t *testing.T, db *mongo.Database) {
 	{
 		// Start Example 46
 
-		projection, err := mongo.Opt.Projection(bson.NewDocument(
+		projection := findopt.Projection(bson.NewDocument(
 			bson.EC.Int32("status", 0),
 			bson.EC.Int32("instock", 0),
 		))
@@ -1258,7 +1258,7 @@ func ProjectionExamples(t *testing.T, db *mongo.Database) {
 			bson.NewDocument(
 				bson.EC.String("status", "A"),
 			),
-			findopt.Projection(projection),
+			projection,
 		)
 
 		// End Example 46
@@ -1287,7 +1287,7 @@ func ProjectionExamples(t *testing.T, db *mongo.Database) {
 	{
 		// Start Example 47
 
-		projection, err := mongo.Opt.Projection(bson.NewDocument(
+		projection := findopt.Projection(bson.NewDocument(
 			bson.EC.Int32("item", 1),
 			bson.EC.Int32("status", 1),
 			bson.EC.Int32("size.uom", 1),
@@ -1299,7 +1299,7 @@ func ProjectionExamples(t *testing.T, db *mongo.Database) {
 			bson.NewDocument(
 				bson.EC.String("status", "A"),
 			),
-			findopt.Projection(projection),
+			projection,
 		)
 
 		// End Example 47
@@ -1333,7 +1333,7 @@ func ProjectionExamples(t *testing.T, db *mongo.Database) {
 	{
 		// Start Example 48
 
-		projection, err := mongo.Opt.Projection(bson.NewDocument(
+		projection := findopt.Projection(bson.NewDocument(
 			bson.EC.Int32("size.uom", 0),
 		))
 		require.NoError(t, err)
@@ -1343,7 +1343,7 @@ func ProjectionExamples(t *testing.T, db *mongo.Database) {
 			bson.NewDocument(
 				bson.EC.String("status", "A"),
 			),
-			findopt.Projection(projection),
+			projection,
 		)
 
 		// End Example 48
@@ -1377,7 +1377,7 @@ func ProjectionExamples(t *testing.T, db *mongo.Database) {
 	{
 		// Start Example 49
 
-		projection, err := mongo.Opt.Projection(bson.NewDocument(
+		projection := findopt.Projection(bson.NewDocument(
 			bson.EC.Int32("item", 1),
 			bson.EC.Int32("status", 1),
 			bson.EC.Int32("instock.qty", 1),
@@ -1389,7 +1389,7 @@ func ProjectionExamples(t *testing.T, db *mongo.Database) {
 			bson.NewDocument(
 				bson.EC.String("status", "A"),
 			),
-			findopt.Projection(projection),
+			projection,
 		)
 
 		// End Example 49
@@ -1435,7 +1435,7 @@ func ProjectionExamples(t *testing.T, db *mongo.Database) {
 	{
 		// Start Example 50
 
-		projection, err := mongo.Opt.Projection(bson.NewDocument(
+		projection := findopt.Projection(bson.NewDocument(
 			bson.EC.Int32("item", 1),
 			bson.EC.Int32("status", 1),
 			bson.EC.SubDocumentFromElements("instock",
@@ -1449,7 +1449,7 @@ func ProjectionExamples(t *testing.T, db *mongo.Database) {
 			bson.NewDocument(
 				bson.EC.String("status", "A"),
 			),
-			findopt.Projection(projection),
+			projection,
 		)
 
 		// End Example 50

mongo/change_stream.go

@@ -14,6 +14,7 @@ import (
 	"github.com/mongodb/mongo-go-driver/bson"
 	"github.com/mongodb/mongo-go-driver/core/command"
 	"github.com/mongodb/mongo-go-driver/core/option"
+	"github.com/mongodb/mongo-go-driver/mongo/changestreamopt"
 )
 
 // ErrMissingResumeToken indicates that a change stream notification from the server did not
@@ -33,16 +34,21 @@ const errorCodeNotMaster int32 = 10107
 const errorCodeCursorNotFound int32 = 43
 
 func newChangeStream(ctx context.Context, coll *Collection, pipeline interface{},
-	opts ...option.ChangeStreamOptioner) (*changeStream, error) {
+	opts ...changestreamopt.ChangeStream) (*changeStream, error) {
 
 	pipelineArr, err := transformAggregatePipeline(pipeline)
 	if err != nil {
 		return nil, err
 	}
 
+	csOpts, err := changestreamopt.BundleChangeStream(opts...).Unbundle(true)
+	if err != nil {
+		return nil, err
+	}
+
 	changeStreamOptions := bson.NewDocument()
 
-	for _, opt := range opts {
+	for _, opt := range csOpts {
 		err = opt.Option(changeStreamOptions)
 		if err != nil {
 			return nil, err
@@ -61,7 +67,7 @@ func newChangeStream(ctx context.Context, coll *Collection, pipeline interface{}
 
 	cs := &changeStream{
 		pipeline: pipelineArr,
-		options:  opts,
+		options:  csOpts,
 		coll:     coll,
 		cursor:   cursor,
 	}
@@ -90,7 +96,7 @@ func (cs *changeStream) Next(ctx context.Context) bool {
 		}
 	}
 
-	resumeToken := Opt.ResumeAfter(cs.resumeToken)
+	resumeToken := changestreamopt.ResumeAfter(cs.resumeToken).ConvertChangeStreamOption()
 	found := false
 
 	for i, opt := range cs.options {