messaging

Documentation

Index

• Constants
• Variables
• func GetMapping() map[string]reflect.Type
• func GetPayloadID(payload any) (uint, bool)
• func GetRegisted() []any
• func PubEntitiesSince(ctx context.Context, keys []string, since time.Time, to time.Time) error
• func QueryEntities[T any](ctx context.Context, db *gorm.DB, logger *zap.Logger, since time.Time, ...) ([]any, error)
• func Reg[T any](payload T)
• 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 DispathFn
• type GormAction
• type GormObjSyncService
  • func NewGormObjSyncService(ms MessagingService, logger *zap.Logger, db *gorm.DB) *GormObjSyncService
  • func (ss *GormObjSyncService) Abandoned(kp *GormPayload, errCode, topic, consumer string)
  • func (ss *GormObjSyncService) ProcessGormObject(ctx context.Context, topic, consumer string, kp *GormPayload, payload any, ...) error
  • func (ss *GormObjSyncService) ReceiveGormObjectSaved(ctx context.Context, topic, consumer string, raw []byte) error
• type GormPayload
  • func ToKeyAndPayload(raw []byte) (*GormPayload, error)
• type MessagingService
• type MessagnePending
• type Processor
• type QueryFn
• type Sharding

Constants

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 (
	AbandonedChan       chan any
	GormCallbackEnabled = true
)

var (
	DefaultGormToipc     = "scm.gorm.saved"
	GormMessagingEnabled = true
)

var ConsumerName string

var ResetTopics []string

var SyncPageSize = 1000

Functions

func GetMapping

func GetMapping() map[string]reflect.Type

func GetPayloadID

func GetPayloadID(payload any) (uint, bool)

try to get payload ID value as uint, return false if payload doesn't have ID field

func GetRegisted

func GetRegisted() []any

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

func PubEntitiesSince

func PubEntitiesSince(ctx context.Context, keys []string, since time.Time, to time.Time) error

func QueryEntities

func QueryEntities[T any](ctx context.Context, db *gorm.DB, logger *zap.Logger, since time.Time, to time.Time, index int, queryDeleted bool) ([]any, error)

func Reg

func Reg[T any](payload T)

should never pass any into this function

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 DispathFn

type DispathFn func(ctx context.Context, topic, consumer string, kp *GormPayload, payload any, tt reflect.Type) error

type GormAction

type GormAction string

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

type GormObjSyncService

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

func NewGormObjSyncService

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

func (*GormObjSyncService) Abandoned

func (ss *GormObjSyncService) Abandoned(kp *GormPayload, errCode, topic, consumer string)

func (*GormObjSyncService) ProcessGormObject

func (ss *GormObjSyncService) ProcessGormObject(ctx context.Context, topic, consumer string, kp *GormPayload, payload any, tt reflect.Type) error

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 string
	SynctAt time.Time
}

func ToKeyAndPayload

func ToKeyAndPayload(raw []byte) (*GormPayload, error)

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 MessagnePending

type MessagnePending struct {
	Topic    string
	Group    string
	Schedule string
	MaxBatch int
	Limit    int
}

type Processor

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

type QueryFn

type QueryFn func(ctx context.Context, db *gorm.DB, logger *zap.Logger, since time.Time, to time.Time, index int, queryDeleted bool) ([]any, error)

type Sharding

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