d0zingcat/gocryptotrader
1
0
Fork 0
mirror of https://github.com/d0zingcat/gocryptotrader.git synced 2026-06-01 07:26:48 +00:00
Code Issues Packages Projects Releases Wiki Activity

egnine/sync-manager: add config support (#1326)

Browse Source 
* allows sync manager customisation for values and logs

* config-example add

* who doesnt like more coverage?

* ensures you can actually disable it via config el oh el

* less ifs, better control

* fix verbose

* sync trades default false

* fix summary being printed when not enabled

* fixes config checker and output

* nits

* I can put this behind me now

* Fixed logCaSiNg

Co-authored-by: Adrian Gallagher <adrian.gallagher@thrasher.io>

* combines if statements

---------

Co-authored-by: Adrian Gallagher <adrian.gallagher@thrasher.io>
This commit is contained in:
Scott
2023-08-29 15:48:41 +10:00
committed by GitHub
parent 04320f7aee
commit d3102a08dc
10 changed files with 281 additions and 108 deletions
Download Patch File Download Diff File Expand all files Collapse all files

121
engine/sync_manager.go
View File

@@ -32,21 +32,15 @@ const (
)
var (
createdCounter = 0
removedCounter = 0
// DefaultSyncerWorkers limits the number of sync workers
DefaultSyncerWorkers = 15
// DefaultSyncerTimeoutREST the default time to switch from REST to websocket protocols without a response
DefaultSyncerTimeoutREST = time.Second * 15
// DefaultSyncerTimeoutWebsocket the default time to switch from websocket to REST protocols without a response
DefaultSyncerTimeoutWebsocket = time.Minute
errNoSyncItemsEnabled = errors.New("no sync items enabled")
errUnknownSyncItem = errors.New("unknown sync item")
errCouldNotSyncNewData = errors.New("could not sync new data")
createdCounter = 0
removedCounter = 0
errNoSyncItemsEnabled = errors.New("no sync items enabled")
errUnknownSyncItem = errors.New("unknown sync item")
errCouldNotSyncNewData = errors.New("could not sync new data")
)
// setupSyncManager starts a new CurrencyPairSyncer
func setupSyncManager(c *SyncManagerConfig, exchangeManager iExchangeManager, remoteConfig *config.RemoteControlConfig, websocketRoutineManagerEnabled bool) (*syncManager, error) {
func setupSyncManager(c *config.SyncManagerConfig, exchangeManager iExchangeManager, remoteConfig *config.RemoteControlConfig, websocketRoutineManagerEnabled bool) (*syncManager, error) {
if c == nil {
return nil, fmt.Errorf("%T %w", c, common.ErrNilPointer)
}
@@ -62,15 +56,15 @@ func setupSyncManager(c *SyncManagerConfig, exchangeManager iExchangeManager, re
}
if c.NumWorkers <= 0 {
c.NumWorkers = DefaultSyncerWorkers
c.NumWorkers = config.DefaultSyncerWorkers
}
if c.TimeoutREST <= time.Duration(0) {
c.TimeoutREST = DefaultSyncerTimeoutREST
c.TimeoutREST = config.DefaultSyncerTimeoutREST
}
if c.TimeoutWebsocket <= time.Duration(0) {
c.TimeoutWebsocket = DefaultSyncerTimeoutWebsocket
c.TimeoutWebsocket = config.DefaultSyncerTimeoutWebsocket
}
if c.FiatDisplayCurrency.IsEmpty() {
@@ -120,6 +114,11 @@ func (m *syncManager) Start() error {
if !atomic.CompareAndSwapInt32(&m.started, 0, 1) {
return ErrSubSystemAlreadyStarted
}
if !m.config.SynchronizeTicker &&
!m.config.SynchronizeOrderbook &&
!m.config.SynchronizeTrades {
return errNoSyncItemsEnabled
}
m.shutdown = make(chan bool)
m.initSyncWG.Add(1)
m.inService.Done()
@@ -196,19 +195,22 @@ func (m *syncManager) Start() error {
}
if atomic.CompareAndSwapInt32(&m.initSyncStarted, 0, 1) {
log.Debugf(log.SyncMgr,
"Exchange CurrencyPairSyncer initial sync started. %d items to process.",
createdCounter)
if m.config.LogInitialSyncEvents {
log.Debugf(log.SyncMgr,
"Exchange CurrencyPairSyncer initial sync started. %d items to process.",
createdCounter)
}
m.initSyncStartTime = time.Now()
}
go func() {
m.initSyncWG.Wait()
if atomic.CompareAndSwapInt32(&m.initSyncCompleted, 0, 1) {
log.Debugf(log.SyncMgr, "Exchange CurrencyPairSyncer initial sync is complete.")
completedTime := time.Now()
log.Debugf(log.SyncMgr, "Exchange CurrencyPairSyncer initial sync took %v [%v sync items].",
completedTime.Sub(m.initSyncStartTime), createdCounter)
if m.config.LogInitialSyncEvents {
log.Debugf(log.SyncMgr, "Exchange CurrencyPairSyncer initial sync is complete.")
log.Debugf(log.SyncMgr, "Exchange CurrencyPairSyncer initial sync took %v [%v sync items].",
time.Since(m.initSyncStartTime), createdCounter)
}
if !m.config.SynchronizeContinuously {
log.Debugln(log.SyncMgr, "Exchange CurrencyPairSyncer stopping.")
@@ -382,13 +384,15 @@ func (m *syncManager) WebsocketUpdate(exchangeName string, p currency.Pair, a as
if !s.IsUsingWebsocket {
s.IsUsingWebsocket = true
s.IsUsingREST = false
log.Warnf(log.SyncMgr,
"%s %s %s: %s Websocket re-enabled, switching from rest to websocket",
c.Exchange,
m.FormatCurrency(c.Pair),
strings.ToUpper(c.AssetType.String()),
syncType,
)
if m.config.LogSwitchProtocolEvents {
log.Warnf(log.SyncMgr,
"%s %s %s: %s Websocket re-enabled, switching from rest to websocket",
c.Exchange,
m.FormatCurrency(c.Pair),
strings.ToUpper(c.AssetType.String()),
syncType,
)
}
}
return m.update(c, syncType, err)
@@ -410,12 +414,14 @@ func (m *syncManager) update(c *currencyPairSyncAgent, syncType syncItemType, er
s.HaveData = true
if atomic.LoadInt32(&m.initSyncCompleted) != 1 && !origHadData {
removedCounter++
log.Debugf(log.SyncMgr, "%s %s sync complete %v [%d/%d].",
c.Exchange,
syncType,
m.FormatCurrency(c.Pair),
removedCounter,
createdCounter)
if m.config.LogInitialSyncEvents {
log.Debugf(log.SyncMgr, "%s %s sync complete %v [%d/%d].",
c.Exchange,
syncType,
m.FormatCurrency(c.Pair),
removedCounter,
createdCounter)
}
m.initSyncWG.Done()
}
@@ -532,13 +538,15 @@ func (m *syncManager) syncTicker(c *currencyPairSyncAgent, e exchange.IBotExchan
// Downgrade to REST
s.IsUsingWebsocket = false
s.IsUsingREST = true
log.Warnf(log.SyncMgr,
"%s %s %s: No ticker update after %s, switching from websocket to rest",
c.Exchange,
m.FormatCurrency(c.Pair),
strings.ToUpper(c.AssetType.String()),
m.config.TimeoutWebsocket,
)
if m.config.LogSwitchProtocolEvents {
log.Warnf(log.SyncMgr,
"%s %s %s: No ticker update after %s, switching from websocket to rest",
c.Exchange,
m.FormatCurrency(c.Pair),
strings.ToUpper(c.AssetType.String()),
m.config.TimeoutWebsocket,
)
}
}
if s.IsUsingREST && time.Since(s.LastUpdated) > m.config.TimeoutREST {
@@ -605,13 +613,15 @@ func (m *syncManager) syncOrderbook(c *currencyPairSyncAgent, e exchange.IBotExc
// Downgrade to REST
s.IsUsingWebsocket = false
s.IsUsingREST = true
log.Warnf(log.SyncMgr,
"%s %s %s: No orderbook update after %s, switching from websocket to rest",
c.Exchange,
m.FormatCurrency(c.Pair).String(),
strings.ToUpper(c.AssetType.String()),
m.config.TimeoutWebsocket,
)
if m.config.LogSwitchProtocolEvents {
log.Warnf(log.SyncMgr,
"%s %s %s: No orderbook update after %s, switching from websocket to rest",
c.Exchange,
m.FormatCurrency(c.Pair).String(),
strings.ToUpper(c.AssetType.String()),
m.config.TimeoutWebsocket,
)
}
}
if s.IsUsingREST && time.Since(s.LastUpdated) > m.config.TimeoutREST {
@@ -691,6 +701,9 @@ func (m *syncManager) PrintTickerSummary(result *ticker.Price, protocol string,
if m == nil || atomic.LoadInt32(&m.started) == 0 {
return
}
if !m.config.SynchronizeTicker {
return
}
if err != nil {
if err == common.ErrNotYetImplemented {
log.Warnf(log.SyncMgr, "Failed to get %s ticker. Error: %s",
@@ -706,6 +719,9 @@ func (m *syncManager) PrintTickerSummary(result *ticker.Price, protocol string,
// ignoring error as not all tickers have volume populated and error is not actionable
_ = stats.Add(result.ExchangeName, result.Pair, result.AssetType, result.Last, result.Volume)
if !m.config.LogSyncUpdateEvents {
return
}
if result.Pair.Quote.IsFiatCurrency() &&
!result.Pair.Quote.Equal(m.fiatDisplayCurrency) &&
@@ -771,6 +787,9 @@ func (m *syncManager) PrintOrderbookSummary(result *orderbook.Base, protocol str
if m == nil || atomic.LoadInt32(&m.started) == 0 {
return
}
if !m.config.SynchronizeOrderbook {
return
}
if err != nil {
if result == nil {
log.Errorf(log.OrderBook, "Failed to get %s orderbook. Error: %s",
@@ -795,7 +814,9 @@ func (m *syncManager) PrintOrderbookSummary(result *orderbook.Base, protocol str
err)
return
}
if !m.config.LogSyncUpdateEvents {
return
}
bidsAmount, bidsValue := result.TotalBidsAmount()
asksAmount, asksValue := result.TotalAsksAmount()