Support bulkwrite

pkg/mongo_client/bulk_write_model_factory.go

@@ -0,0 +1,138 @@
+package mongo_client
+
+import (
+	"fmt"
+	"go.mongodb.org/mongo-driver/bson"
+	"go.mongodb.org/mongo-driver/mongo"
+	"go.mongodb.org/mongo-driver/mongo/options"
+)
+
+func parseModels(arg []map[string][]bson.Raw) []mongo.WriteModel {
+	var models = make([]mongo.WriteModel, 0, len(arg))
+	for _, v := range arg {
+		for kk, vv := range v {
+			models = append(models, makeModel(kk, vv))
+		}
+	}
+	return models
+}
+
+func makeModel(k string, v []bson.Raw) mongo.WriteModel {
+	switch k {
+	case "insertOne":
+		m := mongo.NewInsertOneModel()
+		if len(v) == 0 {
+			return m
+		}
+		m.SetDocument(v[0])
+		return m
+	case "updateOne":
+		m := mongo.NewUpdateOneModel()
+		if len(v) == 0 {
+			return m
+		}
+		m.SetFilter(v[0])
+		if len(v) == 1 {
+			return m
+		}
+		m.SetUpdate(v[1])
+		if len(v) == 2 {
+			return m
+		}
+		o := getOptions(v[2])
+		m.SetUpsert(o.Upsert)
+		if o.Collation == nil {
+			return m
+		}
+		m.SetCollation(o.Collation)
+		return m
+	case "updateMany":
+		m := mongo.NewUpdateManyModel()
+		if len(v) == 0 {
+			return m
+		}
+		m.SetFilter(v[0])
+		if len(v) == 1 {
+			return m
+		}
+		m.SetUpdate(v[1])
+		if len(v) == 2 {
+			return m
+		}
+		o := getOptions(v[2])
+		m.SetUpsert(o.Upsert)
+		if o.Collation == nil {
+			return m
+		}
+		m.SetCollation(o.Collation)
+		return m
+	case "replaceOne":
+		m := mongo.NewReplaceOneModel()
+		if len(v) == 0 {
+			return m
+		}
+		m.SetFilter(v[0])
+		if len(v) == 1 {
+			return m
+		}
+		m.SetReplacement(v[1])
+		if len(v) == 2 {
+			return m
+		}
+		o := getOptions(v[2])
+		m.SetUpsert(o.Upsert)
+		if o.Collation == nil {
+			return m
+		}
+		m.SetCollation(o.Collation)
+		return m
+	case "deleteOne":
+		m := mongo.NewDeleteOneModel()
+		if len(v) == 0 {
+			return m
+		}
+		m.SetFilter(v[0])
+		if len(v) == 1 {
+			return m
+		}
+		o := getOptions(v[1])
+		if o.Collation == nil {
+			return m
+		}
+		m.SetCollation(o.Collation)
+		return m
+	case "deleteMany":
+		m := mongo.NewDeleteManyModel()
+		if len(v) == 0 {
+			return m
+		}
+		m.SetFilter(v[0])
+		if len(v) == 1 {
+			return m
+		}
+		o := getOptions(v[1])
+		if o.Collation == nil {
+			return m
+		}
+		m.SetCollation(o.Collation)
+		return m
+	default:
+		return nil
+	}
+}
+
+type option struct {
+	Collation    *options.Collation   `bson:"collation"`
+	Upsert       bool                 `bson:"upsert"`
+	ArrayFilters options.ArrayFilters `bson:"arrayFilters"`
+}
+
+func getOptions(v bson.Raw) *option {
+	var o option
+	err := bson.Unmarshal(v, &o)
+	if err != nil {
+		panic(err)
+	}
+	fmt.Println(o)
+	return &o
+}

pkg/mongo_client/mongo_client.go

@@ -251,6 +251,23 @@ func (m *MongoProxy) Aggregate(payload []byte, result *[]byte) error {
 	})
 }
 
+type BulkWriteCmd struct {
+	Database   string
+	Collection string
+	Operations []map[string][]bson.Raw
+	Opts       []*options.BulkWriteOptions
+}
+
+func (m *MongoProxy) BulkWrite(payload []byte, result *[]byte) error {
+	cmd := &BulkWriteCmd{}
+	return m.exec(cmd, payload, result, func(ctx context.Context, r *interface{}) (err error) {
+		collection := m.client.Database(cmd.Database).Collection(cmd.Collection)
+		models := parseModels(cmd.Operations)
+		*r, err = collection.BulkWrite(ctx, models, cmd.Opts...)
+		return err
+	})
+}
+
 type DropCmd struct {
 	Database   string
 	Collection string

src/MongoClient/Collection.php

@@ -12,7 +12,11 @@
 namespace Hyperf\GoTask\MongoClient;
 
 use Hyperf\Contract\ConfigInterface;
-use MongoDB\BSON\ObjectId;
+use Hyperf\GoTask\MongoClient\Type\BulkWriteResult;
+use Hyperf\GoTask\MongoClient\Type\DeleteResult;
+use Hyperf\GoTask\MongoClient\Type\InsertManyResult;
+use Hyperf\GoTask\MongoClient\Type\InsertOneResult;
+use Hyperf\GoTask\MongoClient\Type\UpdateResult;
 use function MongoDB\BSON\fromPHP;
 use function MongoDB\BSON\toPHP;
 
@@ -48,24 +52,22 @@ public function __construct(MongoProxy $mongo, ConfigInterface $config, string $
         $this->collection = $collection;
     }
 
-    public function insertOne($document = [], array $opts = []) : ?ObjectId
+    public function insertOne($document = [], array $opts = []): InsertOneResult
     {
         $document = $this->sanitize($document);
         $data = $this->mongo->insertOne($this->makePayload([
             'Record' => $document,
         ], $opts));
-        $typeMap = ['document' => 'array',  'root' => 'array'];
-        return toPHP($data, $typeMap)['insertedid'];
+        return toPHP($data, ['root' => InsertOneResult::class]);
     }
 
-    public function insertMany($documents = [], array $opts = []) : ?array
+    public function insertMany($documents = [], array $opts = []): InsertManyResult
     {
         $documents = $this->sanitize($documents);
         $data = $this->mongo->insertMany($this->makePayload([
             'Records' => $documents,
         ], $opts));
-        $typeMap = ['document' => 'array',  'root' => 'array'];
-        return toPHP($data, $typeMap)['insertedids'];
+        return toPHP($data, ['root' => InsertManyResult::class]);
     }
 
     public function find($filter = [], array $opts = [], $typeMap = ['document' => 'array',  'root' => 'array'])
@@ -74,7 +76,7 @@ public function find($filter = [], array $opts = [], $typeMap = ['document' => '
         $data = $this->mongo->find($this->makePayload([
             'Filter' => $filter,
         ], $opts));
-        return toPHP($data, $typeMap);
+        return $data !== '' ? toPHP($data, $typeMap) : [];
     }
 
     public function findOne($filter = [], array $opts = [], $typeMap = ['document' => 'array',  'root' => 'array'])
@@ -83,43 +85,43 @@ public function findOne($filter = [], array $opts = [], $typeMap = ['document' =
         $data = $this->mongo->findOne($this->makePayload([
             'Filter' => $filter,
         ], $opts));
-        return toPHP($data, $typeMap);
+        return $data !== '' ? toPHP($data, $typeMap): [];
     }
 
-    public function updateOne($filter = [], $update = [], array $opts = [], $typeMap = ['document' => 'array',  'root' => 'array'])
+    public function updateOne($filter = [], $update = [], array $opts = []): UpdateResult
     {
         $filter = $this->sanitize($filter);
         $update = $this->sanitize($update);
         $data = $this->mongo->updateOne($this->makePayload([
             'Filter' => $filter,
             'Update' => $update,
         ], $opts));
-        return toPHP($data, $typeMap);
+        return toPHP($data, ['root' => UpdateResult::class]);
     }
 
-    public function updateMany($filter = [], $update = [], array $opts = [], $typeMap = ['document' => 'array',  'root' => 'array'])
+    public function updateMany($filter = [], $update = [], array $opts = []): UpdateResult
     {
         $filter = $this->sanitize($filter);
         $update = $this->sanitize($update);
         $data = $this->mongo->updateMany($this->makePayload([
             'Filter' => $filter,
             'Update' => $update,
         ], $opts));
-        return toPHP($data, $typeMap);
+        return toPHP($data, ['root' => UpdateResult::class]);
     }
 
-    public function replaceOne($filter = [], $replace = [], array $opts = [], $typeMap = ['document' => 'array',  'root' => 'array'])
+    public function replaceOne($filter = [], $replace = [], array $opts = []): UpdateResult
     {
         $filter = $this->sanitize($filter);
         $replace = $this->sanitize($replace);
         $data = $this->mongo->replaceOne($this->makePayload([
             'Filter' => $filter,
             'Replace' => $replace,
         ], $opts));
-        return toPHP($data, $typeMap);
+        return toPHP($data, ['root' => UpdateResult::class]);
     }
 
-    public function countDocuments($filter = [], array $opts = [], $typeMap = ['document' => 'int'])
+    public function countDocuments($filter = [], array $opts = []): int
     {
         $filter = $this->sanitize($filter);
         $data = $this->mongo->countDocuments($this->makePayload([
@@ -128,22 +130,22 @@ public function countDocuments($filter = [], array $opts = [], $typeMap = ['docu
         return unpack('P', $data)[1];
     }
 
-    public function deleteOne($filter = [], array $opts = [], $typeMap = ['document' => 'array',  'root' => 'array'])
+    public function deleteOne($filter = [], array $opts = []): DeleteResult
     {
         $filter = $this->sanitize($filter);
         $data = $this->mongo->deleteOne($this->makePayload([
             'Filter' => $filter,
         ], $opts));
-        return toPHP($data, $typeMap);
+        return toPHP($data, ['root' => DeleteResult::class]);
     }
 
-    public function deleteMany($filter = [], array $opts = [], $typeMap = ['document' => 'array',  'root' => 'array'])
+    public function deleteMany($filter = [], array $opts = []): DeleteResult
     {
         $filter = $this->sanitize($filter);
         $data = $this->mongo->deleteMany($this->makePayload([
             'Filter' => $filter,
         ], $opts));
-        return toPHP($data, $typeMap);
+        return toPHP($data, ['root' => DeleteResult::class]);
     }
 
     public function aggregate($pipeline = [], array $opts = [], $typeMap = ['document' => 'array',  'root' => 'array'])
@@ -152,7 +154,16 @@ public function aggregate($pipeline = [], array $opts = [], $typeMap = ['documen
         $data = $this->mongo->aggregate($this->makePayload([
             'Pipeline' => $pipeline,
         ], $opts));
-        return toPHP($data, $typeMap);
+        return $data !== '' ? toPHP($data, $typeMap) : [];
+    }
+
+    public function bulkWrite($operations = [], array $opts = []) : BulkWriteResult
+    {
+        $operations = $this->sanitize($operations);
+        $data = $this->mongo->bulkWrite($this->makePayload([
+            'Operations' => $operations,
+        ], $opts));
+        return toPHP($data, ['root' => BulkWriteResult::class]);
     }
 
     public function drop()

src/MongoClient/MongoProxy.php

@@ -85,4 +85,9 @@ public function updateOne(string $payload): string
     {
         return (string) parent::call('MongoProxy.UpdateOne', $payload, GoTask::PAYLOAD_RAW);
     }
+
+    public function bulkWrite(string $payload): string
+    {
+        return (string) parent::call('MongoProxy.BulkWrite', $payload, GoTask::PAYLOAD_RAW);
+    }
 }

src/MongoClient/MongoTrait.php

@@ -11,8 +11,6 @@
  */
 namespace Hyperf\GoTask\MongoClient;
 
-use function MongoDB\BSON\toPHP;
-
 trait MongoTrait
 {
     private function sanitize($input)