Merge pull request globalsign#2 from eaglerayp/feature/bulkUpdateError

Feature/bulk update error

Author: domodwyer

bulk_test.go

@@ -317,6 +317,28 @@ func (s *S) TestBulkUpdate(c *C) {
 	c.Assert(res, DeepEquals, []doc{{10}, {20}, {30}})
 }
 
+func (s *S) TestBulkUpdateOver1000(c *C) {
+	session, err := mgo.Dial("localhost:40001")
+	c.Assert(err, IsNil)
+	defer session.Close()
+
+	coll := session.DB("mydb").C("mycoll")
+
+	bulk := coll.Bulk()
+	for i := 0; i < 1010; i++ {
+		bulk.Insert(M{"n": i})
+	}
+	_, err = bulk.Run()
+	c.Assert(err, IsNil)
+	bulk = coll.Bulk()
+	for i := 0; i < 1010; i++ {
+		bulk.Update(M{"n": i}, M{"$set": M{"m": i}})
+	}
+	// if not handle well, mongo will return error here
+	_, err = bulk.Run()
+	c.Assert(err, IsNil)
+}
+
 func (s *S) TestBulkUpdateError(c *C) {
 	session, err := mgo.Dial("localhost:40001")
 	c.Assert(err, IsNil)

session.go

@@ -4766,6 +4766,58 @@ func (c *Collection) writeOp(op interface{}, ordered bool) (lerr *LastError, err
 			}
 			return &lerr, nil
 		}
+		if updateOp, ok := op.(bulkUpdateOp); ok && len(updateOp) > 1000 {
+			var lerr LastError
+
+			// Maximum batch size is 1000. Must split out in separate operations for compatibility.
+			for i := 0; i < len(updateOp); i += 1000 {
+				l := i + 1000
+				if l > len(updateOp) {
+					l = len(updateOp)
+				}
+
+				oplerr, err := c.writeOpCommand(socket, safeOp, updateOp[i:l], ordered, bypassValidation)
+
+				lerr.N += oplerr.N
+				lerr.modified += oplerr.modified
+				if err != nil {
+					lerr.ecases = append(lerr.ecases, BulkErrorCase{i, err})
+					if ordered {
+						break
+					}
+				}
+			}
+			if len(lerr.ecases) != 0 {
+				return &lerr, lerr.ecases[0].Err
+			}
+			return &lerr, nil
+		}
+		if deleteOps, ok := op.(bulkDeleteOp); ok && len(deleteOps) > 1000 {
+			var lerr LastError
+
+			// Maximum batch size is 1000. Must split out in separate operations for compatibility.
+			for i := 0; i < len(deleteOps); i += 1000 {
+				l := i + 1000
+				if l > len(deleteOps) {
+					l = len(deleteOps)
+				}
+
+				oplerr, err := c.writeOpCommand(socket, safeOp, deleteOps[i:l], ordered, bypassValidation)
+
+				lerr.N += oplerr.N
+				lerr.modified += oplerr.modified
+				if err != nil {
+					lerr.ecases = append(lerr.ecases, BulkErrorCase{i, err})
+					if ordered {
+						break
+					}
+				}
+			}
+			if len(lerr.ecases) != 0 {
+				return &lerr, lerr.ecases[0].Err
+			}
+			return &lerr, nil
+		}
 		return c.writeOpCommand(socket, safeOp, op, ordered, bypassValidation)
 	} else if updateOps, ok := op.(bulkUpdateOp); ok {
 		var lerr LastError