1st
This commit is contained in:
270
daemon/daemon.go
Normal file
270
daemon/daemon.go
Normal file
@@ -0,0 +1,270 @@
|
||||
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")
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user