onexbet/daemon/daemon.go

package daemon

/*
Отписка от матча:
- метод Unwatch
- когда матч исчезнет из лайва
*/

import (
	"fmt"
	"log"
	"sync"
	"time"

	"gordenko.dev/dima/onexbet"
	"gordenko.dev/dima/ws"
	"gordenko.dev/dima/ws/publisher"
)

const (
	defaultLiveMatchesLoadInterval = 60 * time.Second
	defaultPollingInterval         = 5 * time.Second
)

type Options struct {
	Logger                  *log.Logger
	LiveMatchesLoadInterval int
	PollingInterval         int
	SportIDs                []onexbet.OnexbetSport
	WsPublisher             *publisher.PublicPublisher
}

type Daemon struct {
	mutex                   sync.Mutex
	publisher               *publisher.PublicPublisher
	port                    int
	liveMatchesLoadInterval time.Duration
	logger                  *log.Logger
	//liveMatches             map[string]onexbet.LiveMatch
	teamLiveMatches map[onexbet.OnexbetSport][]onexbet.LiveMatch
	// matchID => isClosed flag
	pollingMatches  map[onexbet.OnexbetSport]map[string]*bool
	pollingInterval time.Duration
	sportIDs        []onexbet.OnexbetSport
}

func New(opt Options) *Daemon {
	if opt.Logger == nil {
		panic("Logger option is required")
	}
	s := &Daemon{
		logger:          opt.Logger,
		publisher:       opt.WsPublisher,
		pollingMatches:  make(map[onexbet.OnexbetSport]map[string]*bool),
		teamLiveMatches: make(map[onexbet.OnexbetSport][]onexbet.LiveMatch),
		sportIDs:        opt.SportIDs,
	}
	if opt.LiveMatchesLoadInterval <= 0 {
		s.liveMatchesLoadInterval = defaultLiveMatchesLoadInterval
	} else {
		s.liveMatchesLoadInterval = time.Duration(opt.LiveMatchesLoadInterval) * time.Second
	}
	if opt.PollingInterval <= 0 {
		s.pollingInterval = defaultPollingInterval
	} else {
		s.pollingInterval = time.Duration(opt.PollingInterval) * time.Second
	}
	for _, sportID := range s.sportIDs {
		s.pollingMatches[sportID] = make(map[string]*bool)
	}
	return s
}

func (s *Daemon) OnWsClientConnected(conn *ws.Conn) {
	s.mutex.Lock()
	defer s.mutex.Unlock()

	for _, sportID := range s.sportIDs {
		conn.Send(onexbet.WsMsgLiveMatches, s.getLiveMatchesMessage(sportID))
	}

	s.publisher.SubscribeTo(conn, onexbet.TopicLive)
}

func (s *Daemon) getLiveMatchesMessage(sportID onexbet.OnexbetSport) onexbet.LiveMatchesMessage {
	var watching []string

	for matchID := range s.pollingMatches[sportID] {
		watching = append(watching, matchID)
	}

	return onexbet.LiveMatchesMessage{
		SportID:  sportID,
		Matches:  s.teamLiveMatches[sportID],
		Watching: watching,
	}
}

func (s *Daemon) loadLiveMatches(sportID onexbet.OnexbetSport) {
	s.logger.Printf("Load live matches\n")

	list, err := onexbet.ListLiveMatchesBySport(sportID)
	if err != nil {
		s.logger.Printf("ListLiveMatchesBySport(sportID=%d): %s\n", sportID, err)
		return
	}
	//s.logger.Printf("NewMatches: %# v\n", pretty.Formatter(newMatches))
	//s.logger.Printf("NewMatches: %d\n", len(newMatchesList))

	// sync current live matches with new matches

	s.mutex.Lock()
	defer s.mutex.Unlock()

	s.teamLiveMatches[sportID] = list

	s.publisher.Publish(onexbet.TopicLive, onexbet.WsMsgLiveMatches, s.getLiveMatchesMessage(sportID))
}

// Метод может прислать обновление для матча после закрытия. Это не проблема.
func (s *Daemon) pollTeamMatch(m map[string]*bool, req onexbet.WatchReq, isClosedPtr *bool) { //closeCh chan struct{}) {
	s.logger.Printf("poll team match %s\n", req.MatchID)
	for {
		if *isClosedPtr {
			s.logger.Printf("match %s polling is closed by command\n", req.MatchID)
			return
		}

		data, isMatchFinished, err := onexbet.LoadTeamLiveMatchData(req.MatchID)
		if err != nil {
			s.logger.Printf("LoadTeamLiveMatchData: %s; matchID=%s", err, req.MatchID)
		} else {
			//s.sendLiveMatchData(data)
			//s.logger.Printf("match update: %v\n\n", data)
			// Мутекс нужен, чтобы защитить Publisher, ибо он НЕ ThreadSafe!

			if isMatchFinished {
				// Отправлять сообщение не нужно, ибо Tipper обнаружит что матч исчез из
				// polling и запросит для него результат
				//s.mutex.Lock()
				//delete(m, req.MatchID)
				//s.publisher.Publish(ChannelLive, onexbet.WSMessageMatchFinished, req)
				//s.mutex.Unlock()
				// выходим
				s.logger.Printf("match %s polling is finished\n", req.MatchID)
				return
			}

			s.mutex.Lock()
			s.publisher.Publish(onexbet.TopicLive, onexbet.WsMsgTeamLiveMatchData, data)
			s.mutex.Unlock()
		}

		time.Sleep(s.pollingInterval)
	}
}

// Метод может прислать обновление для матча после закрытия. Это не проблема.
func (s *Daemon) pollTennisMatch(m map[string]*bool, req onexbet.WatchReq, isClosedPtr *bool) { //closeCh chan struct{}) {
	s.logger.Printf("poll tennis match %s\n", req.MatchID)
	for {
		if *isClosedPtr {
			s.logger.Printf("match %s polling is closed by command\n", req.MatchID)
			return
		}

		data, isMatchFinished, err := onexbet.LoadTennisLiveMatchData(req.MatchID)
		if err != nil {
			s.logger.Printf("LoadTennisLiveMatchData: %s; matchID=%s", err, req.MatchID)
		} else {
			//s.sendLiveMatchData(data)
			//s.logger.Printf("match update: %v\n\n", data)
			// Мутекс нужен, чтобы защитить Publisher, ибо он НЕ ThreadSafe!

			if isMatchFinished {
				// Отправлять сообщение не нужно, ибо Tipper обнаружит что матч исчез из
				// polling и запросит для него результат
				//s.mutex.Lock()
				//delete(m, req.MatchID)
				//s.publisher.Publish(ChannelLive, onexbet.WSMessageMatchFinished, req)
				//s.mutex.Unlock()
				// выходим
				s.logger.Printf("match %s polling is finished\n", req.MatchID)
				return
			}

			s.mutex.Lock()
			s.publisher.Publish(onexbet.TopicLive, onexbet.WsMsgTennisLiveMatchData, data)
			s.mutex.Unlock()
		}

		time.Sleep(s.pollingInterval)
	}
}

func (s *Daemon) Watch(req onexbet.WatchReq) (err error) {
	s.logger.Printf("Watch: sport=%d, matchID=%s\n", req.SportID, req.MatchID)

	s.mutex.Lock()
	defer s.mutex.Unlock()

	m, ok := s.pollingMatches[req.SportID]
	if !ok {
		err = fmt.Errorf("Unknown sportID: %d", req.SportID)
		return
	}

	_, ok = m[req.MatchID]
	if ok {
		// already polling
		s.logger.Printf("match %s already polling\n", req.MatchID)
		return
	}

	var isClosed bool

	isClosedPtr := &isClosed

	m[req.MatchID] = isClosedPtr //closeCh

	if req.SportID == onexbet.Tennis {
		go s.pollTennisMatch(m, req, isClosedPtr)
	} else {
		go s.pollTeamMatch(m, req, isClosedPtr)
	}
	return
}

func (s *Daemon) Unwatch(req onexbet.WatchReq) (err error) {
	s.logger.Printf("Unwatch: sport=%d, matchID=%s\n", req.SportID, req.MatchID)

	s.mutex.Lock()
	defer s.mutex.Unlock()

	m, ok := s.pollingMatches[req.SportID]
	if !ok {
		err = fmt.Errorf("Unknown sportID: %d", req.SportID)
		return
	}

	isClosedPtr, ok := m[req.MatchID]
	if !ok {
		return
	}
	// Сигнализируем горутине чтобы она закрылась
	*isClosedPtr = true
	delete(m, req.MatchID)
	return
}

func (s *Daemon) Run() {
	for _, sportID := range s.sportIDs {
		s.loadLiveMatches(sportID)
		s.logger.Println("load end")
	}

	s.logger.Printf("liveMatchesLoadInterval: %d\n", int(s.liveMatchesLoadInterval.Seconds()))
	ticker := time.NewTicker(s.liveMatchesLoadInterval)

	for {
		s.logger.Println("new loop")
		select {
		case <-ticker.C:
			for _, sportID := range s.sportIDs {
				s.loadLiveMatches(sportID)
				s.logger.Println("load end")
			}
		}
	}
}