messaging

Documentation

Index

• Constants
• Variables
• func GetRegisted() []any
• func PubGormDeleted(ctx context.Context, payload any) error
• func PubGormSaved(ctx context.Context, payload any) error
• func Reg(payload any)
• type DefaultMessgingService
  • func (msg *DefaultMessgingService) ProcessPendings(ctx context.Context, topic, group string, processor Processor)
  • func (msg *DefaultMessgingService) Pub(ctx context.Context, topic string, payload any) error
  • func (msg *DefaultMessgingService) Sub(ctx context.Context, topic, group string, processor Processor) error
• type GormAction
• type GormObjSyncService
  • func NewGormObjSyncService(ms MessagingService, logger *zap.Logger, db *gorm.DB) *GormObjSyncService
  • func (ss *GormObjSyncService) ReceiveGormObjectSaved(ctx context.Context, topic, consumer string, raw []byte) error
• type GormPayload
• type IDbase
• type MessagingService
• type Processor
• type Sharding

Constants

const (
	GormActionSave   = "save"
	GormActionDelete = "delete"
)

const (
	DefaultMsgLimit          = math.MaxInt16
	DefaultAttKey            = "payload"
	DefaultSchedule          = "@every 30m"
	DefaultDeadLetterDurtion = 8 * time.Hour //if messaging pending for more than this duration, will be put to dead letter
)

Variables

var DefaultGormToipc = "scm.gorm.saved"

var GormMessagingEnabled = true

Functions

func GetRegisted

func GetRegisted() []any

GetRegisted function is used to obtain a list of registered entities.

func PubGormDeleted

func PubGormDeleted(ctx context.Context, payload any) error

func PubGormSaved

func PubGormSaved(ctx context.Context, payload any) error

func Reg

func Reg(payload any)

Types

type DefaultMessgingService

type DefaultMessgingService struct {
	Logger          *zap.Logger
	Client          *redis.Client
	PendingSchedule string
	Settings        map[string]int64 // settings for streaming limit settings. default 10000
}

func (*DefaultMessgingService) ProcessPendings

func (msg *DefaultMessgingService) ProcessPendings(ctx context.Context, topic, group string, processor Processor)

func (*DefaultMessgingService) Pub

func (msg *DefaultMessgingService) Pub(ctx context.Context, topic string, payload any) error

func (*DefaultMessgingService) Sub

func (msg *DefaultMessgingService) Sub(ctx context.Context, topic, group string, processor Processor) error

type GormAction

type GormAction string

type GormObjSyncService

type GormObjSyncService struct {
	MessageService MessagingService
	DB             *gorm.DB
	Logger         *zap.Logger
	Sharding       Sharding
}

func NewGormObjSyncService

func NewGormObjSyncService(ms MessagingService, logger *zap.Logger, db *gorm.DB) *GormObjSyncService

func (*GormObjSyncService) ReceiveGormObjectSaved

func (ss *GormObjSyncService) ReceiveGormObjectSaved(ctx context.Context, topic, consumer string, raw []byte) error

type GormPayload

type GormPayload struct {
	Key     string
	Action  GormAction
	Payload []byte
}

type IDbase

type IDbase interface {
	GetID() uint
}

type MessagingService

type MessagingService interface {
	Pub(ctx context.Context, topic string, payload any) error
	Sub(ctx context.Context, topic, consumer string, processor Processor) error
}

MessagingService， default impl Redis streaming.

type Processor

type Processor func(ctx context.Context, topic, consumer string, payload []byte) error

type Sharding

type Sharding func(tx *gorm.DB, payload any) (tablename string, err error)