qdy пре 2 месеци
родитељ
комит
bb2474c47f
1 измењених фајлова са 56 додато и 40 уклоњено
  1. 56
    40
      factory/mongodb/mongodb_factory.go

+ 56
- 40
factory/mongodb/mongodb_factory.go Прегледај датотеку

8
 	"sync"
8
 	"sync"
9
 	"time"
9
 	"time"
10
 
10
 
11
+	"git.x2erp.com/qdy/go-base/config"
11
 	"git.x2erp.com/qdy/go-base/config/subconfigs"
12
 	"git.x2erp.com/qdy/go-base/config/subconfigs"
12
 	"git.x2erp.com/qdy/go-base/logger"
13
 	"git.x2erp.com/qdy/go-base/logger"
13
 	"go.mongodb.org/mongo-driver/bson"
14
 	"go.mongodb.org/mongo-driver/bson"
21
 	client *mongo.Client
22
 	client *mongo.Client
22
 	db     *mongo.Database
23
 	db     *mongo.Database
23
 	config *subconfigs.MongoDBConfig
24
 	config *subconfigs.MongoDBConfig
24
-	//mu     sync.RWMutex // 添加读写锁保护
25
+	mu     sync.RWMutex // 添加读写锁保护
25
 }
26
 }
26
 
27
 
27
 var (
28
 var (
30
 	initErr             error
31
 	initErr             error
31
 )
32
 )
32
 
33
 
33
-// GetMongoDBFactory 获取MongoDB工厂单例
34
-func GetMongoDBFactory(config *subconfigs.MongoDBConfig) *MongoDBFactory {
34
+// CreateFactory 获取MongoDB工厂单例
35
+func CreateFactory(cfg config.IConfig) *MongoDBFactory {
36
+
37
+	config := cfg.GetMongoDBConfig()
35
 
38
 
36
 	instanceMongodbOnce.Do(func() {
39
 	instanceMongodbOnce.Do(func() {
37
 		if config == nil {
40
 		if config == nil {
40
 
43
 
41
 		// 设置默认值
44
 		// 设置默认值
42
 		if config.Timeout == 0 {
45
 		if config.Timeout == 0 {
43
-			config.Timeout = 20 * time.Second
46
+			config.Timeout = 30
44
 		}
47
 		}
45
 		if config.MaxPoolSize == 0 {
48
 		if config.MaxPoolSize == 0 {
46
 			config.MaxPoolSize = 100
49
 			config.MaxPoolSize = 100
66
 
69
 
67
 		// 创建客户端选项
70
 		// 创建客户端选项
68
 		clientOptions := options.Client().
71
 		clientOptions := options.Client().
69
-			ApplyURI(config.URI).
70
-			SetConnectTimeout(config.Timeout).
71
-			SetSocketTimeout(config.Timeout).
72
-			SetServerSelectionTimeout(config.Timeout).
73
-			SetMaxPoolSize(config.MaxPoolSize).
74
-			SetMinPoolSize(config.MinPoolSize).
75
-			SetMaxConnIdleTime(5 * time.Minute).
76
-			SetHeartbeatInterval(10 * time.Second)
72
+			ApplyURI(config.URI)
73
+			//SetConnectTimeout(config.Timeout).
74
+			//SetSocketTimeout(config.Timeout).
75
+			//SetServerSelectionTimeout(config.Timeout).
76
+			//SetMaxPoolSize(config.MaxPoolSize).
77
+			//SetMinPoolSize(config.MinPoolSize).
78
+			//SetMaxConnIdleTime(5 * time.Minute).
79
+			//SetHeartbeatInterval(10 * time.Second)
77
 
80
 
78
 		// 设置认证
81
 		// 设置认证
79
 		if config.Username != "" && config.Password != "" {
82
 		if config.Username != "" && config.Password != "" {
92
 		}
95
 		}
93
 
96
 
94
 		// 创建上下文
97
 		// 创建上下文
95
-		ctx, cancel := context.WithTimeout(context.Background(), config.Timeout)
98
+		ctx, cancel := context.WithTimeout(context.Background(), config.GetTimeout())
96
 		defer cancel()
99
 		defer cancel()
97
 
100
 
98
 		log.Printf(" context.WithTimeout ... successfully.")
101
 		log.Printf(" context.WithTimeout ... successfully.")
133
 	return instanceMongodb
136
 	return instanceMongodb
134
 }
137
 }
135
 
138
 
139
+func (c *MongoDBFactory) GetTimeout() time.Duration {
140
+	return c.config.GetTimeout()
141
+}
142
+
143
+// TestConnection 测试连接
144
+func (f *MongoDBFactory) TestConnection() {
145
+	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
146
+	defer cancel()
147
+
148
+	if err := f.client.Ping(ctx, readpref.Primary()); err != nil {
149
+		log.Printf("MongoDB连接测试失败: %v", err)
150
+	} else {
151
+		log.Printf("MongoDB连接测试通过.")
152
+	}
153
+
154
+}
155
+
136
 // ========== MongoDBFactory 实例方法 ==========
156
 // ========== MongoDBFactory 实例方法 ==========
137
 
157
 
138
 // GetClient 获取MongoDB客户端
158
 // GetClient 获取MongoDB客户端
150
 	return f.db.Collection(collectionName)
170
 	return f.db.Collection(collectionName)
151
 }
171
 }
152
 
172
 
173
+func (f *MongoDBFactory) GetName() string {
174
+	return "MongoDBFactory"
175
+}
176
+
153
 // Close 关闭MongoDB连接
177
 // Close 关闭MongoDB连接
154
 func (f *MongoDBFactory) Close() {
178
 func (f *MongoDBFactory) Close() {
155
 	if f.client != nil {
179
 	if f.client != nil {
174
 	return *f.config
198
 	return *f.config
175
 }
199
 }
176
 
200
 
177
-// TestConnection 测试连接
178
-func (f *MongoDBFactory) TestConnection() error {
179
-	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
180
-	defer cancel()
181
-
182
-	return f.client.Ping(ctx, readpref.Primary())
183
-}
184
-
185
 // ========== 快捷操作方法(增强版) ==========
201
 // ========== 快捷操作方法(增强版) ==========
186
 
202
 
187
 // InsertOne 插入单个文档,返回是否成功
203
 // InsertOne 插入单个文档,返回是否成功
188
 func (f *MongoDBFactory) InsertOne(collectionName string, document interface{}) bool {
204
 func (f *MongoDBFactory) InsertOne(collectionName string, document interface{}) bool {
189
 	collection := f.GetCollection(collectionName)
205
 	collection := f.GetCollection(collectionName)
190
-	ctx, cancel := context.WithTimeout(context.Background(), f.config.Timeout)
206
+	ctx, cancel := context.WithTimeout(context.Background(), f.config.GetTimeout())
191
 	defer cancel()
207
 	defer cancel()
192
 
208
 
193
 	_, err := collection.InsertOne(ctx, document)
209
 	_, err := collection.InsertOne(ctx, document)
201
 // InsertOneWithResult 插入单个文档并返回结果
217
 // InsertOneWithResult 插入单个文档并返回结果
202
 func (f *MongoDBFactory) InsertOneWithResult(collectionName string, document interface{}) (*mongo.InsertOneResult, bool) {
218
 func (f *MongoDBFactory) InsertOneWithResult(collectionName string, document interface{}) (*mongo.InsertOneResult, bool) {
203
 	collection := f.GetCollection(collectionName)
219
 	collection := f.GetCollection(collectionName)
204
-	ctx, cancel := context.WithTimeout(context.Background(), f.config.Timeout)
220
+	ctx, cancel := context.WithTimeout(context.Background(), f.config.GetTimeout())
205
 	defer cancel()
221
 	defer cancel()
206
 
222
 
207
 	result, err := collection.InsertOne(ctx, document)
223
 	result, err := collection.InsertOne(ctx, document)
215
 // InsertMany 插入多个文档,返回是否成功
231
 // InsertMany 插入多个文档,返回是否成功
216
 func (f *MongoDBFactory) InsertMany(collectionName string, documents []interface{}) bool {
232
 func (f *MongoDBFactory) InsertMany(collectionName string, documents []interface{}) bool {
217
 	collection := f.GetCollection(collectionName)
233
 	collection := f.GetCollection(collectionName)
218
-	ctx, cancel := context.WithTimeout(context.Background(), f.config.Timeout)
234
+	ctx, cancel := context.WithTimeout(context.Background(), f.config.GetTimeout())
219
 	defer cancel()
235
 	defer cancel()
220
 
236
 
221
 	_, err := collection.InsertMany(ctx, documents)
237
 	_, err := collection.InsertMany(ctx, documents)
229
 // InsertManyWithResult 插入多个文档并返回结果
245
 // InsertManyWithResult 插入多个文档并返回结果
230
 func (f *MongoDBFactory) InsertManyWithResult(collectionName string, documents []interface{}) (*mongo.InsertManyResult, bool) {
246
 func (f *MongoDBFactory) InsertManyWithResult(collectionName string, documents []interface{}) (*mongo.InsertManyResult, bool) {
231
 	collection := f.GetCollection(collectionName)
247
 	collection := f.GetCollection(collectionName)
232
-	ctx, cancel := context.WithTimeout(context.Background(), f.config.Timeout)
248
+	ctx, cancel := context.WithTimeout(context.Background(), f.config.GetTimeout())
233
 	defer cancel()
249
 	defer cancel()
234
 
250
 
235
 	result, err := collection.InsertMany(ctx, documents)
251
 	result, err := collection.InsertMany(ctx, documents)
243
 // FindOne 查询单个文档,返回解码后的对象
259
 // FindOne 查询单个文档,返回解码后的对象
244
 func (f *MongoDBFactory) FindOne(collectionName string, filter interface{}, result interface{}) error {
260
 func (f *MongoDBFactory) FindOne(collectionName string, filter interface{}, result interface{}) error {
245
 	collection := f.GetCollection(collectionName)
261
 	collection := f.GetCollection(collectionName)
246
-	ctx, cancel := context.WithTimeout(context.Background(), f.config.Timeout)
262
+	ctx, cancel := context.WithTimeout(context.Background(), f.config.GetTimeout())
247
 	defer cancel()
263
 	defer cancel()
248
 
264
 
249
 	err := collection.FindOne(ctx, filter).Decode(result)
265
 	err := collection.FindOne(ctx, filter).Decode(result)
259
 // FindOneAndDecode 查询单个文档并自动解码(简化版)
275
 // FindOneAndDecode 查询单个文档并自动解码(简化版)
260
 func (f *MongoDBFactory) FindOneAndDecode(collectionName string, filter interface{}) (interface{}, error) {
276
 func (f *MongoDBFactory) FindOneAndDecode(collectionName string, filter interface{}) (interface{}, error) {
261
 	collection := f.GetCollection(collectionName)
277
 	collection := f.GetCollection(collectionName)
262
-	ctx, cancel := context.WithTimeout(context.Background(), f.config.Timeout)
278
+	ctx, cancel := context.WithTimeout(context.Background(), f.config.GetTimeout())
263
 	defer cancel()
279
 	defer cancel()
264
 
280
 
265
 	var result map[string]interface{}
281
 	var result map[string]interface{}
276
 // Find 查询多个文档,返回对象数组
292
 // Find 查询多个文档,返回对象数组
277
 func (f *MongoDBFactory) Find(collectionName string, filter interface{}, results interface{}, opts ...*options.FindOptions) error {
293
 func (f *MongoDBFactory) Find(collectionName string, filter interface{}, results interface{}, opts ...*options.FindOptions) error {
278
 	collection := f.GetCollection(collectionName)
294
 	collection := f.GetCollection(collectionName)
279
-	ctx, cancel := context.WithTimeout(context.Background(), f.config.Timeout)
295
+	ctx, cancel := context.WithTimeout(context.Background(), f.config.GetTimeout())
280
 	defer cancel()
296
 	defer cancel()
281
 
297
 
282
 	cursor, err := collection.Find(ctx, filter, opts...)
298
 	cursor, err := collection.Find(ctx, filter, opts...)
302
 // UpdateOneWithResult 更新单个文档并返回结果
318
 // UpdateOneWithResult 更新单个文档并返回结果
303
 func (f *MongoDBFactory) UpdateOneWithResult(collectionName string, filter interface{}, update interface{}) (*mongo.UpdateResult, bool) {
319
 func (f *MongoDBFactory) UpdateOneWithResult(collectionName string, filter interface{}, update interface{}) (*mongo.UpdateResult, bool) {
304
 	collection := f.GetCollection(collectionName)
320
 	collection := f.GetCollection(collectionName)
305
-	ctx, cancel := context.WithTimeout(context.Background(), f.config.Timeout)
321
+	ctx, cancel := context.WithTimeout(context.Background(), f.config.GetTimeout())
306
 	defer cancel()
322
 	defer cancel()
307
 
323
 
308
 	result, err := collection.UpdateOne(ctx, filter, update)
324
 	result, err := collection.UpdateOne(ctx, filter, update)
316
 // CountDocuments 统计文档数量,返回数量
332
 // CountDocuments 统计文档数量,返回数量
317
 func (f *MongoDBFactory) CountDocuments(collectionName string, filter interface{}) (int64, error) {
333
 func (f *MongoDBFactory) CountDocuments(collectionName string, filter interface{}) (int64, error) {
318
 	collection := f.GetCollection(collectionName)
334
 	collection := f.GetCollection(collectionName)
319
-	ctx, cancel := context.WithTimeout(context.Background(), f.config.Timeout)
335
+	ctx, cancel := context.WithTimeout(context.Background(), f.config.GetTimeout())
320
 	defer cancel()
336
 	defer cancel()
321
 
337
 
322
 	count, err := collection.CountDocuments(ctx, filter)
338
 	count, err := collection.CountDocuments(ctx, filter)
330
 // Aggregate 聚合查询,返回对象数组
346
 // Aggregate 聚合查询,返回对象数组
331
 func (f *MongoDBFactory) Aggregate(collectionName string, pipeline interface{}, results interface{}) error {
347
 func (f *MongoDBFactory) Aggregate(collectionName string, pipeline interface{}, results interface{}) error {
332
 	collection := f.GetCollection(collectionName)
348
 	collection := f.GetCollection(collectionName)
333
-	ctx, cancel := context.WithTimeout(context.Background(), f.config.Timeout)
349
+	ctx, cancel := context.WithTimeout(context.Background(), f.config.GetTimeout())
334
 	defer cancel()
350
 	defer cancel()
335
 
351
 
336
 	cursor, err := collection.Aggregate(ctx, pipeline)
352
 	cursor, err := collection.Aggregate(ctx, pipeline)
350
 // FindOneAndUpdate 查找并更新,返回更新后的文档
366
 // FindOneAndUpdate 查找并更新,返回更新后的文档
351
 func (f *MongoDBFactory) FindOneAndUpdate(collectionName string, filter interface{}, update interface{}, result interface{}) error {
367
 func (f *MongoDBFactory) FindOneAndUpdate(collectionName string, filter interface{}, update interface{}, result interface{}) error {
352
 	collection := f.GetCollection(collectionName)
368
 	collection := f.GetCollection(collectionName)
353
-	ctx, cancel := context.WithTimeout(context.Background(), f.config.Timeout)
369
+	ctx, cancel := context.WithTimeout(context.Background(), f.config.GetTimeout())
354
 	defer cancel()
370
 	defer cancel()
355
 
371
 
356
 	err := collection.FindOneAndUpdate(ctx, filter, update).Decode(result)
372
 	err := collection.FindOneAndUpdate(ctx, filter, update).Decode(result)
366
 // FindOneAndDelete 查找并删除,返回删除的文档
382
 // FindOneAndDelete 查找并删除,返回删除的文档
367
 func (f *MongoDBFactory) FindOneAndDelete(collectionName string, filter interface{}, result interface{}) error {
383
 func (f *MongoDBFactory) FindOneAndDelete(collectionName string, filter interface{}, result interface{}) error {
368
 	collection := f.GetCollection(collectionName)
384
 	collection := f.GetCollection(collectionName)
369
-	ctx, cancel := context.WithTimeout(context.Background(), f.config.Timeout)
385
+	ctx, cancel := context.WithTimeout(context.Background(), f.config.GetTimeout())
370
 	defer cancel()
386
 	defer cancel()
371
 
387
 
372
 	err := collection.FindOneAndDelete(ctx, filter).Decode(result)
388
 	err := collection.FindOneAndDelete(ctx, filter).Decode(result)
382
 // BulkWrite 批量写入操作,返回是否成功
398
 // BulkWrite 批量写入操作,返回是否成功
383
 func (f *MongoDBFactory) BulkWrite(collectionName string, operations []mongo.WriteModel) bool {
399
 func (f *MongoDBFactory) BulkWrite(collectionName string, operations []mongo.WriteModel) bool {
384
 	collection := f.GetCollection(collectionName)
400
 	collection := f.GetCollection(collectionName)
385
-	ctx, cancel := context.WithTimeout(context.Background(), f.config.Timeout)
401
+	ctx, cancel := context.WithTimeout(context.Background(), f.config.GetTimeout())
386
 	defer cancel()
402
 	defer cancel()
387
 
403
 
388
 	_, err := collection.BulkWrite(ctx, operations)
404
 	_, err := collection.BulkWrite(ctx, operations)
396
 // CreateIndex 创建索引,返回是否成功
412
 // CreateIndex 创建索引,返回是否成功
397
 func (f *MongoDBFactory) CreateIndex(collectionName string, keys interface{}, opts ...*options.IndexOptions) bool {
413
 func (f *MongoDBFactory) CreateIndex(collectionName string, keys interface{}, opts ...*options.IndexOptions) bool {
398
 	collection := f.GetCollection(collectionName)
414
 	collection := f.GetCollection(collectionName)
399
-	ctx, cancel := context.WithTimeout(context.Background(), f.config.Timeout)
415
+	ctx, cancel := context.WithTimeout(context.Background(), f.config.GetTimeout())
400
 	defer cancel()
416
 	defer cancel()
401
 
417
 
402
 	indexModel := mongo.IndexModel{
418
 	indexModel := mongo.IndexModel{
478
 // UpdateOne 更新单个文档,返回是否执行成功和影响记录数
494
 // UpdateOne 更新单个文档,返回是否执行成功和影响记录数
479
 func (f *MongoDBFactory) UpdateOne(collectionName string, filter interface{}, update interface{}) (bool, int64) {
495
 func (f *MongoDBFactory) UpdateOne(collectionName string, filter interface{}, update interface{}) (bool, int64) {
480
 	collection := f.GetCollection(collectionName)
496
 	collection := f.GetCollection(collectionName)
481
-	ctx, cancel := context.WithTimeout(context.Background(), f.config.Timeout)
497
+	ctx, cancel := context.WithTimeout(context.Background(), f.config.GetTimeout())
482
 	defer cancel()
498
 	defer cancel()
483
 
499
 
484
 	result, err := collection.UpdateOne(ctx, filter, update)
500
 	result, err := collection.UpdateOne(ctx, filter, update)
500
 // UpdateMany 更新多个文档,返回是否执行成功和影响记录数
516
 // UpdateMany 更新多个文档,返回是否执行成功和影响记录数
501
 func (f *MongoDBFactory) UpdateMany(collectionName string, filter interface{}, update interface{}) (bool, int64) {
517
 func (f *MongoDBFactory) UpdateMany(collectionName string, filter interface{}, update interface{}) (bool, int64) {
502
 	collection := f.GetCollection(collectionName)
518
 	collection := f.GetCollection(collectionName)
503
-	ctx, cancel := context.WithTimeout(context.Background(), f.config.Timeout)
519
+	ctx, cancel := context.WithTimeout(context.Background(), f.config.GetTimeout())
504
 	defer cancel()
520
 	defer cancel()
505
 
521
 
506
 	result, err := collection.UpdateMany(ctx, filter, update)
522
 	result, err := collection.UpdateMany(ctx, filter, update)
515
 // DeleteOne 删除单个文档,返回是否执行成功和删除的记录数
531
 // DeleteOne 删除单个文档,返回是否执行成功和删除的记录数
516
 func (f *MongoDBFactory) DeleteOne(collectionName string, filter interface{}) (bool, int64) {
532
 func (f *MongoDBFactory) DeleteOne(collectionName string, filter interface{}) (bool, int64) {
517
 	collection := f.GetCollection(collectionName)
533
 	collection := f.GetCollection(collectionName)
518
-	ctx, cancel := context.WithTimeout(context.Background(), f.config.Timeout)
534
+	ctx, cancel := context.WithTimeout(context.Background(), f.config.GetTimeout())
519
 	defer cancel()
535
 	defer cancel()
520
 
536
 
521
 	result, err := collection.DeleteOne(ctx, filter)
537
 	result, err := collection.DeleteOne(ctx, filter)
537
 // DeleteMany 删除多个文档,返回是否执行成功和删除的记录数
553
 // DeleteMany 删除多个文档,返回是否执行成功和删除的记录数
538
 func (f *MongoDBFactory) DeleteMany(collectionName string, filter interface{}) (bool, int64) {
554
 func (f *MongoDBFactory) DeleteMany(collectionName string, filter interface{}) (bool, int64) {
539
 	collection := f.GetCollection(collectionName)
555
 	collection := f.GetCollection(collectionName)
540
-	ctx, cancel := context.WithTimeout(context.Background(), f.config.Timeout)
556
+	ctx, cancel := context.WithTimeout(context.Background(), f.config.GetTimeout())
541
 	defer cancel()
557
 	defer cancel()
542
 
558
 
543
 	result, err := collection.DeleteMany(ctx, filter)
559
 	result, err := collection.DeleteMany(ctx, filter)
552
 // UpsertOne 更新或插入文档(upsert操作)
568
 // UpsertOne 更新或插入文档(upsert操作)
553
 func (f *MongoDBFactory) UpsertOne(collectionName string, filter interface{}, update interface{}) (bool, interface{}) {
569
 func (f *MongoDBFactory) UpsertOne(collectionName string, filter interface{}, update interface{}) (bool, interface{}) {
554
 	collection := f.GetCollection(collectionName)
570
 	collection := f.GetCollection(collectionName)
555
-	ctx, cancel := context.WithTimeout(context.Background(), f.config.Timeout)
571
+	ctx, cancel := context.WithTimeout(context.Background(), f.config.GetTimeout())
556
 	defer cancel()
572
 	defer cancel()
557
 
573
 
558
 	// 设置 upsert 选项
574
 	// 设置 upsert 选项

Loading…
Откажи
Сачувај