pusher/pushers/controller.go

package pushers

import (
	"context"
	"fmt"
	"sync"

	"gitea.timerzz.com/kedaya_haitao/pusher/biz/dal/postgres"
	"gitea.timerzz.com/kedaya_haitao/pusher/kitex_gen/config"
	"github.com/bytedance/sonic"
	"golang.org/x/exp/maps"
	"golang.org/x/sync/errgroup"
)

func Init() {
	c = NewController()
}

func Push(ctx context.Context, ids []int64, title, content string) error {
	return c.Push(ctx, ids, title, content)
}

type Pusher interface {
	Push(ctx context.Context, title, content string) error
}

type Controller struct {
	pushers map[int64]Pusher
	l       sync.RWMutex
}

func NewController() *Controller {
	var pushers []config.PusherConfig
	_ = postgres.DB.Find(&pushers)
	ctl := &Controller{
		pushers: make(map[int64]Pusher, len(pushers)),
	}
	for _, pusher := range pushers {
		ctl.pushers[pusher.Id] = NewPusher(pusher)
	}
	fmt.Printf("共%d个pusher\n", len(pushers))
	return ctl
}

var c *Controller

func (c *Controller) Push(ctx context.Context, ids []int64, title, content string) error {
	wg, sub := errgroup.WithContext(ctx)
	if len(ids) == 0 {
		c.l.RLock()
		ids = maps.Keys(c.pushers)
		c.l.RUnlock()
	}
	for _, id := range ids {
		var i = id
		wg.Go(func() error {
			if pusher, ok := c.pushers[i]; ok {
				return pusher.Push(sub, title, content)
			} else {
				var cfg config.PusherConfig
				if err := postgres.DB.Where("id = ?", i).Find(&cfg).Error; err != nil {
					return err
				}
				c.l.Lock()
				c.pushers[cfg.Id] = NewPusher(cfg)
				c.l.Unlock()
				return c.pushers[cfg.Id].Push(sub, title, content)
			}
		})
	}
	return wg.Wait()
}

func NewPusher(cfg config.PusherConfig) Pusher {
	switch cfg.Type {
	case config.PusherConfigType_AnPush:
		var opt config.AnPush
		_ = sonic.UnmarshalString(cfg.Option, &opt)
		return NewAnPush(&opt)
	case config.PusherConfigType_Email:
		var opt config.EmailPush
		_ = sonic.UnmarshalString(cfg.Option, &opt)
		return NewEmailPusher(&opt)
	}
	return nil
}