Skip to content

Commit 2f303f9

Browse files
ReasnoReasno
authored
V2.2.0 release (#18)
Refactor mongodb package to use mongo-ext for maximum compatibility. Upgrade to hyperf 2.0
1 parent 674f52a commit 2f303f9

24 files changed

+1422
-345
lines changed

‎.travis.yml

+1
Original file line numberDiff line numberDiff line change
@@ -26,6 +26,7 @@ before_install:
2626
install:
2727
- cd $TRAVIS_BUILD_DIR
2828
- bash ./tests/swoole.install.sh
29+
- pecl install mongodb || echo "extension=mongodb.so" >> `php --ini | grep "Loaded Configuration" | sed -e "s|.*:\s*||"`
2930
- phpenv config-rm xdebug.ini || echo "xdebug not available"
3031
- phpenv config-add ./tests/ci.ini
3132

‎composer.json

+9-7
Original file line numberDiff line numberDiff line change
@@ -20,17 +20,19 @@
2020
"require": {
2121
"php": ">=7.2",
2222
"ext-swoole": ">=4.4",
23-
"hyperf/pool": "^1.1",
24-
"hyperf/process": "^1.1",
23+
"hyperf/pool": "~2.0.0",
24+
"hyperf/process": "~2.0.0",
2525
"spiral/goridge": "^2.4.1",
26-
"symfony/event-dispatcher": "~4.4.0"
26+
"symfony/event-dispatcher": "^5.1"
2727
},
2828
"require-dev": {
29+
"ext-mongodb": "*",
2930
"friendsofphp/php-cs-fixer": "^2.14",
30-
"hyperf/command": "^1.1",
31-
"hyperf/config": "^1.1",
32-
"hyperf/di": "^1.1",
33-
"hyperf/testing": "1.1.*",
31+
"hyperf/command": "~2.0.0",
32+
"hyperf/config": "~2.0.0",
33+
"hyperf/di": "~2.0.0",
34+
"hyperf/framework": "~2.0.0",
35+
"hyperf/testing": "~2.0.0",
3436
"mockery/mockery": "^1.3",
3537
"phpstan/phpstan": "^0.12",
3638
"swoole/ide-helper": "^4.5"

‎example/sidecar.go

+1-2
Original file line numberDiff line numberDiff line change
@@ -4,10 +4,9 @@ import (
44
"bytes"
55
"encoding/base64"
66
"fmt"
7+
"github.com/hyperf/gotask/v2/pkg/gotask"
78
"io/ioutil"
89
"log"
9-
10-
"github.com/hyperf/gotask/v2/pkg/gotask"
1110
)
1211

1312
// App sample

‎go.mod

+1-2
Original file line numberDiff line numberDiff line change
@@ -6,7 +6,6 @@ require (
66
github.com/fatih/pool v3.0.0+incompatible
77
github.com/oklog/run v1.1.0
88
github.com/pkg/errors v0.9.1
9-
github.com/reasno/gotask v1.0.2
10-
github.com/spiral/goridge/v2 v2.3.3
119
go.mongodb.org/mongo-driver v1.3.3
10+
github.com/spiral/goridge/v2 v2.4.4
1211
)

‎pkg/gotask/middleware.go

+31
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,31 @@
1+
package gotask
2+
3+
import (
4+
"fmt"
5+
)
6+
7+
type Handler func(cmd interface{}, result *interface{}) error
8+
9+
type Middleware func(next Handler) Handler
10+
11+
func Chain(outer Middleware, others ...Middleware) Middleware {
12+
return func(next Handler) Handler {
13+
for i := len(others) - 1; i >= 0; i-- { // reverse
14+
next = others[i](next)
15+
}
16+
return outer(next)
17+
}
18+
}
19+
20+
func PanicRecover() Middleware {
21+
return func(next Handler) Handler {
22+
return func(cmd interface{}, r *interface{}) (e error) {
23+
defer func() {
24+
if rec := recover(); rec != nil {
25+
e = fmt.Errorf("panic: %s", rec)
26+
}
27+
}()
28+
return next(cmd, r)
29+
}
30+
}
31+
}

‎pkg/mongo_client/bulk_write_model_factory.go

+136
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,136 @@
1+
package mongo_client
2+
3+
import (
4+
"go.mongodb.org/mongo-driver/bson"
5+
"go.mongodb.org/mongo-driver/mongo"
6+
"go.mongodb.org/mongo-driver/mongo/options"
7+
)
8+
9+
func parseModels(arg []map[string][]bson.Raw) []mongo.WriteModel {
10+
var models = make([]mongo.WriteModel, 0, len(arg))
11+
for _, v := range arg {
12+
for kk, vv := range v {
13+
models = append(models, makeModel(kk, vv))
14+
}
15+
}
16+
return models
17+
}
18+
19+
func makeModel(k string, v []bson.Raw) mongo.WriteModel {
20+
switch k {
21+
case "insertOne":
22+
m := mongo.NewInsertOneModel()
23+
if len(v) == 0 {
24+
return m
25+
}
26+
m.SetDocument(v[0])
27+
return m
28+
case "updateOne":
29+
m := mongo.NewUpdateOneModel()
30+
if len(v) == 0 {
31+
return m
32+
}
33+
m.SetFilter(v[0])
34+
if len(v) == 1 {
35+
return m
36+
}
37+
m.SetUpdate(v[1])
38+
if len(v) == 2 {
39+
return m
40+
}
41+
o := getOptions(v[2])
42+
m.SetUpsert(o.Upsert)
43+
if o.Collation == nil {
44+
return m
45+
}
46+
m.SetCollation(o.Collation)
47+
return m
48+
case "updateMany":
49+
m := mongo.NewUpdateManyModel()
50+
if len(v) == 0 {
51+
return m
52+
}
53+
m.SetFilter(v[0])
54+
if len(v) == 1 {
55+
return m
56+
}
57+
m.SetUpdate(v[1])
58+
if len(v) == 2 {
59+
return m
60+
}
61+
o := getOptions(v[2])
62+
m.SetUpsert(o.Upsert)
63+
if o.Collation == nil {
64+
return m
65+
}
66+
m.SetCollation(o.Collation)
67+
return m
68+
case "replaceOne":
69+
m := mongo.NewReplaceOneModel()
70+
if len(v) == 0 {
71+
return m
72+
}
73+
m.SetFilter(v[0])
74+
if len(v) == 1 {
75+
return m
76+
}
77+
m.SetReplacement(v[1])
78+
if len(v) == 2 {
79+
return m
80+
}
81+
o := getOptions(v[2])
82+
m.SetUpsert(o.Upsert)
83+
if o.Collation == nil {
84+
return m
85+
}
86+
m.SetCollation(o.Collation)
87+
return m
88+
case "deleteOne":
89+
m := mongo.NewDeleteOneModel()
90+
if len(v) == 0 {
91+
return m
92+
}
93+
m.SetFilter(v[0])
94+
if len(v) == 1 {
95+
return m
96+
}
97+
o := getOptions(v[1])
98+
if o.Collation == nil {
99+
return m
100+
}
101+
m.SetCollation(o.Collation)
102+
return m
103+
case "deleteMany":
104+
m := mongo.NewDeleteManyModel()
105+
if len(v) == 0 {
106+
return m
107+
}
108+
m.SetFilter(v[0])
109+
if len(v) == 1 {
110+
return m
111+
}
112+
o := getOptions(v[1])
113+
if o.Collation == nil {
114+
return m
115+
}
116+
m.SetCollation(o.Collation)
117+
return m
118+
default:
119+
return nil
120+
}
121+
}
122+
123+
type option struct {
124+
Collation *options.Collation `bson:"collation"`
125+
Upsert bool `bson:"upsert"`
126+
ArrayFilters options.ArrayFilters `bson:"arrayFilters"`
127+
}
128+
129+
func getOptions(v bson.Raw) *option {
130+
var o option
131+
err := bson.Unmarshal(v, &o)
132+
if err != nil {
133+
panic(err)
134+
}
135+
return &o
136+
}

‎pkg/mongo_client/config.go

+4
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,7 @@
11
package mongo_client
22

33
import (
4+
"flag"
45
"os"
56
"time"
67
)
@@ -32,6 +33,9 @@ func getTimeout(env string, fallback time.Duration) (result time.Duration) {
3233
// LoadConfig loads Configurations from environmental variables or config file in PHP.
3334
// Environmental variables takes priority.
3435
func LoadConfig() Config {
36+
if !flag.Parsed() {
37+
flag.Parse()
38+
}
3539
return Config{
3640
*globalMongoUri,
3741
*globalMongoConnectTimeout,

‎pkg/mongo_client/middleware.go

+85
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,85 @@
1+
package mongo_client
2+
3+
import (
4+
"encoding/binary"
5+
"fmt"
6+
"github.com/hyperf/gotask/v2/pkg/gotask"
7+
"github.com/pkg/errors"
8+
"go.mongodb.org/mongo-driver/bson"
9+
"go.mongodb.org/mongo-driver/mongo"
10+
)
11+
12+
// BsonDeserialize deserializes bson cmd into a struct cmd
13+
func BsonDeserialize(ex interface{}) gotask.Middleware {
14+
return func(next gotask.Handler) gotask.Handler {
15+
return func(cmd interface{}, r *interface{}) error {
16+
b, ok := cmd.([]byte)
17+
if !ok {
18+
return fmt.Errorf("bsonDeserialize only accepts []byte")
19+
}
20+
e := bson.Unmarshal(b, ex)
21+
if e != nil {
22+
return errors.Wrap(e, "fails to unmarshal bson")
23+
}
24+
return next(ex, r)
25+
}
26+
}
27+
}
28+
29+
// BsonSerialize serializes any result into a bson encoded result
30+
func BsonSerialize() gotask.Middleware {
31+
return func(next gotask.Handler) gotask.Handler {
32+
return func(cmd interface{}, r *interface{}) (e error) {
33+
defer func() {
34+
if e != nil {
35+
*r = []byte{}
36+
return
37+
}
38+
if *r == nil {
39+
*r = []byte{}
40+
return
41+
}
42+
switch (*r).(type) {
43+
case int64:
44+
b := make([]byte, 8)
45+
binary.LittleEndian.PutUint64(b, uint64((*r).(int64)))
46+
*r = b
47+
return
48+
case string:
49+
*r = []byte((*r).(string))
50+
return
51+
default:
52+
_, *r, e = bson.MarshalValue(r)
53+
if e != nil {
54+
e = errors.Wrap(e, "unable to serialize bson")
55+
}
56+
}
57+
58+
}()
59+
return next(cmd, r)
60+
}
61+
}
62+
}
63+
64+
func ErrorFilter() gotask.Middleware {
65+
return func(next gotask.Handler) gotask.Handler {
66+
return func(cmd interface{}, r *interface{}) (e error) {
67+
defer func() {
68+
if e == mongo.ErrNilCursor || e == mongo.ErrNilDocument {
69+
e = nil
70+
}
71+
e = errors.Wrap(e, "error while executing mongo command")
72+
}()
73+
return next(cmd, r)
74+
}
75+
}
76+
}
77+
78+
func stackMiddleware(ex interface{}) gotask.Middleware {
79+
return gotask.Chain(
80+
gotask.PanicRecover(),
81+
BsonDeserialize(ex),
82+
BsonSerialize(),
83+
ErrorFilter(),
84+
)
85+
}

0 commit comments

Comments
 (0)