add namespace toString method

model/event.go

@@ -1,6 +1,8 @@
 package model
 
 import (
+	"fmt"
+
 	"go.mongodb.org/mongo-driver/bson"
 	"go.mongodb.org/mongo-driver/bson/primitive"
 )
@@ -28,3 +30,7 @@ type Namespace struct {
 	Database   string `bson:"db,omitempty" json:"db,omitempty"`
 	Collection string `bson:"coll,omitempty" json:"coll,omitempty"`
 }
+
+func (ns Namespace) String() string {
+	return fmt.Sprintf("%s.%s", ns.Database, ns.Collection)
+}

operator/mongo/mongo.go

@@ -46,27 +46,24 @@ func (m *mongoOperator) Replace(e model.ChangeEvent) error {
 }
 
 func (m *mongoOperator) Update(e model.ChangeEvent) error {
-	if e.FullDocument == nil {
-		update := bson.M{}
+	update := bson.M{}
 
-		if len(e.UpdateDescription.UpdatedFields) != 0 {
-			update["$set"] = e.UpdateDescription.UpdatedFields
-		}
+	if len(e.UpdateDescription.UpdatedFields) != 0 {
+		update["$set"] = e.UpdateDescription.UpdatedFields
+	}
 
-		if len(e.UpdateDescription.RemovedFields) != 0 {
-			unset := bson.M{}
-			for _, field := range e.UpdateDescription.RemovedFields {
-				unset[field] = ""
-			}
-			update["$unset"] = unset
+	if len(e.UpdateDescription.RemovedFields) != 0 {
+		unset := bson.M{}
+		for _, field := range e.UpdateDescription.RemovedFields {
+			unset[field] = ""
 		}
-
-		_, err := m.Database(e.Namespace.Database).
-			Collection(e.Namespace.Collection).
-			UpdateOne(context.Background(), e.DocumentKey, update)
-		return err
+		update["$unset"] = unset
 	}
-	return m.Replace(e)
+
+	_, err := m.Database(e.Namespace.Database).
+		Collection(e.Namespace.Collection).
+		UpdateOne(context.Background(), e.DocumentKey, update)
+	return err
 }
 
 func (m *mongoOperator) Drop(e model.ChangeEvent) error {
@@ -76,8 +73,8 @@ func (m *mongoOperator) Drop(e model.ChangeEvent) error {
 }
 
 func (m *mongoOperator) Rename(e model.ChangeEvent) error {
-	from := e.Namespace.Database + "." + e.Namespace.Collection
-	to := e.To.Database + "." + e.To.Collection
+	from := e.Namespace.String()
+	to := e.To.String()
 
 	result := m.Database("admin").
 		RunCommand(