Commit 83fdf144 authored by zhengyaoqiu's avatar zhengyaoqiu

feat(colorway): 代码优化

parent 5dfde887
...@@ -11,23 +11,28 @@ import ( ...@@ -11,23 +11,28 @@ import (
"sync" "sync"
) )
type Cache[T any] interface { type ChangeStream[T any] interface {
ChangeStreamFilter(data ChangeStreamData) bson.M
DocumentEqual(a, b T) bool DocumentEqual(a, b T) bool
Filter(data ChangeStreamData) bson.M
}
type Cache[T any] interface {
GetCacheDocuments() []T GetCacheDocuments() []T
StartCache(ctx context.Context) error
} }
type Base[T any] struct { type Base[T any] struct {
baseModel model.Model[T] baseModel model.Model[T]
changeStream ChangeStream[T]
documents []T documents []T
rwMutex *sync.RWMutex rwMutex *sync.RWMutex
} }
func NewBase[T any](baseModel model.Model[T]) *Base[T] { func NewBase[T any](baseModel model.Model[T], changeStream ChangeStream[T]) *Base[T] {
return &Base[T]{baseModel: baseModel} return &Base[T]{baseModel: baseModel, changeStream: changeStream}
} }
func (receiver *Base[T]) StartCache(ctx context.Context, cache Cache[T]) error { func (receiver *Base[T]) StartCache(ctx context.Context) error {
documents, _, err := receiver.baseModel.Find(ctx, bson.M{}) documents, _, err := receiver.baseModel.Find(ctx, bson.M{})
if err != nil { if err != nil {
return err return err
...@@ -48,7 +53,7 @@ func (receiver *Base[T]) StartCache(ctx context.Context, cache Cache[T]) error { ...@@ -48,7 +53,7 @@ func (receiver *Base[T]) StartCache(ctx context.Context, cache Cache[T]) error {
log.Println(err) log.Println(err)
continue continue
} }
filter := cache.ChangeStreamFilter(changeStreamData) filter := receiver.changeStream.Filter(changeStreamData)
document, err := receiver.baseModel.FindOne(ctx, filter) document, err := receiver.baseModel.FindOne(ctx, filter)
if err != nil { if err != nil {
log.Println(err) log.Println(err)
...@@ -56,7 +61,7 @@ func (receiver *Base[T]) StartCache(ctx context.Context, cache Cache[T]) error { ...@@ -56,7 +61,7 @@ func (receiver *Base[T]) StartCache(ctx context.Context, cache Cache[T]) error {
} }
documents := receiver.GetCacheDocuments() documents := receiver.GetCacheDocuments()
for i := range documents { for i := range documents {
if cache.DocumentEqual(documents[i], document) { if receiver.changeStream.DocumentEqual(documents[i], document) {
documents[i] = document documents[i] = document
} }
} }
...@@ -86,7 +91,7 @@ func (receiver *Base[T]) setAllDocuments(documents []T) { ...@@ -86,7 +91,7 @@ func (receiver *Base[T]) setAllDocuments(documents []T) {
receiver.documents = documents receiver.documents = documents
} }
func (receiver *Base[T]) ChangeStreamFilter(data ChangeStreamData) bson.M { func (receiver *Base[T]) Filter(data ChangeStreamData) bson.M {
id, err := primitive.ObjectIDFromHex(data.DocumentKey.Id.Oid) id, err := primitive.ObjectIDFromHex(data.DocumentKey.Id.Oid)
if err != nil { if err != nil {
return nil return nil
......
package mongo package mongo
import ( import (
"context"
productModel "git.chillcy.com/golang/colorway/internal/pkg/database/mongo/model/product" productModel "git.chillcy.com/golang/colorway/internal/pkg/database/mongo/model/product"
sitePlatformPriceConfigModel "git.chillcy.com/golang/colorway/internal/pkg/database/mongo/model/site_platform_price_config" sitePlatformPriceConfigModel "git.chillcy.com/golang/colorway/internal/pkg/database/mongo/model/site_platform_price_config"
) )
...@@ -18,3 +19,11 @@ func NewManager(mongoUrl string) *Manager { ...@@ -18,3 +19,11 @@ func NewManager(mongoUrl string) *Manager {
SitePlatformPriceConfigModel: sitePlatformPriceConfigModel.NewSitePlatformPriceConfigModel(mongoUrl, DatabaseColorWay, sitePlatformPriceConfigModel.Collection), SitePlatformPriceConfigModel: sitePlatformPriceConfigModel.NewSitePlatformPriceConfigModel(mongoUrl, DatabaseColorWay, sitePlatformPriceConfigModel.Collection),
} }
} }
func (receiver *Manager) StartCache() error {
err := receiver.SitePlatformPriceConfigModel.StartCache(context.Background())
if err != nil {
return err
}
}
...@@ -13,6 +13,7 @@ type ( ...@@ -13,6 +13,7 @@ type (
SitePlatformPriceConfigModel interface { SitePlatformPriceConfigModel interface {
sitePlatformPriceConfigModel sitePlatformPriceConfigModel
cache.Cache[SitePlatformPriceConfig] cache.Cache[SitePlatformPriceConfig]
cache.ChangeStream[SitePlatformPriceConfig]
} }
customSitePlatformPriceConfigModel struct { customSitePlatformPriceConfigModel struct {
...@@ -25,10 +26,11 @@ type ( ...@@ -25,10 +26,11 @@ type (
func NewSitePlatformPriceConfigModel(url, db, collection string) SitePlatformPriceConfigModel { func NewSitePlatformPriceConfigModel(url, db, collection string) SitePlatformPriceConfigModel {
conn := mon.MustNewModel(url, db, collection) conn := mon.MustNewModel(url, db, collection)
defaultModel := newDefaultSitePlatformPriceConfigModel(conn) defaultModel := newDefaultSitePlatformPriceConfigModel(conn)
return &customSitePlatformPriceConfigModel{ customModel := &customSitePlatformPriceConfigModel{
defaultSitePlatformPriceConfigModel: defaultModel, defaultSitePlatformPriceConfigModel: defaultModel,
Base: cache.NewBase[SitePlatformPriceConfig](defaultModel),
} }
customModel.Base = cache.NewBase[SitePlatformPriceConfig](defaultModel, customModel)
return customModel
} }
func (receiver *customSitePlatformPriceConfigModel) DocumentEqual(a, b SitePlatformPriceConfig) bool { func (receiver *customSitePlatformPriceConfigModel) DocumentEqual(a, b SitePlatformPriceConfig) bool {
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment